목차
- 문제와 접근
- @tool 도구에 HTTP API 연결
- fetch_todos, count_completed_todos 도구
- 전체 데이터 수집 패턴 복습
- 마크다운 표 생성 함수
- build_user_profile 고도화
- build_team_report
- BaseTool로 HTTP 도구 만들기
- StockTool — 스키마 결합 예시
- 오케스트레이터 패턴 (api_tools)
- 핵심 개념 총정리
- 자주 나오는 실수/주의사항
- 부분 성공을 결과에 표현하기
1. 문제와 접근
여러 HTTP 호출로 하나의 결과를 만들면 부분 실패를 어떻게 표현할지가 중요한 설계 문제가 된다. 도구 래퍼와 리포트 함수를 구성하면서 수집, 집계, 표현의 책임을 나누어 본다. 예외를 처리했다는 사실과 결과가 완전하다는 판단은 구분한다.
1
2
3
4
5
6
7
8
9
10
| import os, json, time, requests
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool, StructuredTool
from langchain_core.messages import HumanMessage, ToolMessage
from pydantic import BaseModel, Field
load_dotenv()
BASE_URL = "https://jsonplaceholder.typicode.com"
llm = ChatOpenAI(model='gpt-4o-mini')
|
기본 패턴 (fetch_user, fetch_user_post)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| @tool
def fetch_user(user_id):
"""지정된 ID의 사용자 정보를 조회합니다"""
r = requests.get(f"{BASE_URL}/users/{user_id}")
if not r.ok:
return f"error: status {r.status_code}"
data = r.json()
return json.dumps({
'name' : data['name'],
'email' : data['email'],
'phone' : data['phone'],
'company': data['company']['name']
}, ensure_ascii=False)
@tool
def fetch_user_post(user_id):
"""지정된 ID의 사용자가 작성한 게시글을 조회합니다"""
r = requests.get(f"{BASE_URL}/posts", params={'userId': user_id})
if not r.ok:
return f"error: status {r.status_code}"
posts = r.json()
return json.dumps(
[{'id': p['id'], 'title': p['title']} for p in posts],
ensure_ascii=False
)
|
도구 실행 흐름
1
2
3
4
5
6
7
8
9
10
11
12
13
| llm_api = llm.bind_tools([fetch_user, fetch_user_post])
tool_map = {'fetch_user': fetch_user, 'fetch_user_post': fetch_user_post}
messages = [HumanMessage(content='1번 유저 정보와 그 사람이 쓴 게시글 갯수 알려줘')]
resp = llm_api.invoke(messages)
messages.append(resp)
for tc in resp.tool_calls:
result = tool_map[tc['name']].invoke(tc['args'])
messages.append(ToolMessage(content=result, tool_call_id=tc['id']))
final = llm_api.invoke(messages)
print(final.content)
|
3. fetch_todos, count_completed_todos 도구
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| @tool
def fetch_todos(user_id):
"""지정된 사용자의 todo 목록 상위 5개를 조회합니다"""
r = requests.get(f"{BASE_URL}/todos", params={'userId': user_id})
if not r.ok:
return f"error: {r.status_code}"
todos = r.json()[:5]
return json.dumps(todos, ensure_ascii=False)
@tool
def count_completed_todos(user_id):
"""지정된 사용자의 완료된 todo 개수와 전체 개수를 계산합니다"""
r = requests.get(f"{BASE_URL}/todos", params={'userId': user_id})
if not r.ok:
return f"error: {r.status_code}"
todos = r.json()
completed = sum(1 for t in todos if t['completed'])
total = len(todos)
return json.dumps({
'user_id' : user_id,
'completed' : completed,
'total' : total,
'completion_rate': round(completed / total * 100, 1) if total else 0,
}, ensure_ascii=False)
|
실행 결과
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| todo_tools = [fetch_todos, count_completed_todos]
llm_todo = llm.bind_tools(todo_tools)
tool_map = {t.name: t for t in todo_tools}
messages = [HumanMessage(content='1번 유저의 todo 완료율을 알려줘')]
resp = llm_todo.invoke(messages)
messages.append(resp)
for tc in resp.tool_calls:
result = tool_map[tc['name']].invoke(tc['args'])
messages.append(ToolMessage(content=result, tool_call_id=tc['id']))
final = llm_todo.invoke(messages)
print(final.content)
# → "1번 유저의 todo 완료율은 45%입니다. 총 200개 중 90개가 완료되었습니다."
|
4. 전체 데이터 수집 패턴 복습
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| def fetch_all_posts(page_size=10, max_pages=100):
page = 1
while page <= max_pages:
r = requests.get(f"{BASE_URL}/posts",
params={"_page": page, "_limit": page_size})
items = r.json()
if not items:
break
print(f" page {page}: {len(items)}개")
for item in items:
yield item
if len(items) < page_size:
break
page += 1
all_posts = list(fetch_all_posts(page_size=20))
len(all_posts) # → 100
|
완료된 todos만 수집
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| def collect_completed_todos(max_items=20, page_size=10):
collected = []
page = 1
while len(collected) < max_items:
r = requests.get(f"{BASE_URL}/todos",
params={"_page": page, "_limit": page_size})
items = r.json()
if not items:
break
completed = [t for t in items if t.get('completed')]
collected.extend(completed[:max_items - len(collected)])
print(f" page {page}: {len(items)}개, 수집: {len(collected)}개")
if len(items) < page_size:
break
page += 1
return collected[:max_items]
|
5. 마크다운 표 생성 함수
중첩 딕셔너리 경로를 점(.)으로 접근하는 유틸 + 마크다운 표
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| def get_by_path(data, path, default=None):
current = data
for key in path.split("."):
if isinstance(current, dict) and key in current:
current = current[key]
else:
return default
return current
def users_to_markdown(user_ids: list, fields: list):
rows = []
for uid in user_ids:
r = requests.get(f"{BASE_URL}/users/{uid}")
if not r.ok:
continue
data = r.json()
row = {f: get_by_path(data, f, "N/A") for f in fields}
rows.append(row)
# 헤더 행
header = "| " + " | ".join(fields) + " |"
# 구분선
sep = "|" + "|".join(["-" * (len(f) + 2) for f in fields]) + "|"
# 데이터 행
body_lines = [
"| " + " | ".join(str(row[f]) for f in fields) + " |"
for row in rows
]
return '\n'.join([header, sep] + body_lines)
table = users_to_markdown(
user_ids=[1, 2, 3],
fields=['name', 'email', 'address.city']
)
print(table)
|
출력:
1
2
3
4
5
| | name | email | address.city |
|------|-------|--------------|
| Leanne Graham | Sincere@april.biz | Gwenborough |
| Ervin Howell | Shanna@melissa.tv | Wisokyburgh |
| ... | ... | ... |
|
6. build_user_profile 고도화
하나의 호출이 실패해도 나머지를 반환하는 방식은 유용하지만, 누락된 항목이 없는 값인지 수집하지 못한 값인지 표시해야 한다. 아래 예제를 확장할 때는 HTTP 실패 상태도 기록하고, 일부 요청의 실패가 집계값을 정상적인 0으로 보이게 만들지 않는지 확인한다.
각 API 호출을 try-except로 감싸서 하나가 실패해도 나머지 정보는 수집한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
| def build_user_profile(user_id):
profile = {'user_id': user_id}
try:
r = requests.get(f"{BASE_URL}/users/{user_id}")
if r.ok:
u = r.json()
profile['name'] = u['name']
profile['email'] = u['email']
profile['city'] = u['address']['city']
except:
profile['user_error'] = 'error'
try:
r = requests.get(f"{BASE_URL}/posts", params={'userId': user_id})
if r.ok:
posts = r.json()
profile['post_count'] = len(posts)
profile['recent_posts'] = [p['title'][:40] for p in posts[:3]]
except Exception as e:
profile['post_error'] = str(e)
try:
r = requests.get(f"{BASE_URL}/todos", params={'userId': user_id})
if r.ok:
todos = r.json()
profile['todo_total'] = len(todos)
profile['todo_completed'] = sum(1 for t in todos if t['completed'])
except:
profile['todo_error'] = 'error'
return profile
build_user_profile(1)
# → {'user_id': 1, 'name': 'Leanne Graham', 'email': '...', 'city': 'Gwenborough',
# 'post_count': 10, 'recent_posts': [...], 'todo_total': 20, 'todo_completed': 11}
|
7. build_team_report
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| def build_team_report(user_list):
members = []
for uid in user_list:
try:
u = requests.get(f"{BASE_URL}/users/{uid}")
p = requests.get(f"{BASE_URL}/posts", params={'userId': uid})
t = requests.get(f"{BASE_URL}/todos", params={'userId': uid})
if not (u.ok and p.ok and t.ok):
continue
members.append({
'user_id' : uid,
'name' : u.json()['name'],
'post_count': len(p.json()),
})
except:
continue
total_posts = sum(m['post_count'] for m in members)
top = max(members, key=lambda m: m['post_count']) if members else None
return {
'team_size' : len(members),
'members' : members,
'top_poster': {'user_id': top['user_id'], 'post_count': top['post_count']} if top else None,
}
build_team_report([1, 2, 3, 4, 5])
|
캐시가 필요할 때 BaseTool을 상속한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| from langchain_core.tools import BaseTool
class APICallerTool(BaseTool):
name: str = "api_caller"
description: str = "REST API를 호출하여 데이터를 조회합니다"
call_count: int = 0
cache: dict = {}
def _run(self, endpoint: str) -> str:
if endpoint in self.cache:
return f"[cache] {self.cache[endpoint]}"
self.call_count += 1
data = {
"users/1": {"name": "abcd", "email": "abcd@example.com"},
"posts/1": {"title": "1st post", "userId": 1},
}
result = json.dumps(data.get(endpoint, {"error": "not found"}), ensure_ascii=False)
self.cache[endpoint] = result
return f"[API #{self.call_count}] {result}"
api_tool = APICallerTool()
api_tool.invoke({'endpoint': 'users/1'}) # → [API #1] ...
api_tool.invoke({'endpoint': 'users/1'}) # → [cache] ... (두 번째부터 캐시)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| from typing import Type
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
class StockInput(BaseModel):
ticker: str = Field(description="주식 티커 (예: AAPL)")
include_history: bool = Field(default=False, description="과거 데이터 포함 여부")
class StockTool(BaseTool):
name: str = "stock_lookup"
description: str = "주식 정보를 조회합니다"
args_schema: Type[BaseModel] = StockInput
query_log: list = []
def _run(self, ticker: str, include_history: bool = False) -> str:
prices = {"AAPL": 100.5, "GOOGL": 102.1, "TSLA": 280.3}
price = prices.get(ticker.upper())
if not price:
return f"{ticker} no data"
self.query_log.append({"ticker": ticker, "time": time.time()})
result = f"{ticker}: ${price}"
if include_history:
result += f" (30일 변동: +{price * 0.05:.2f})"
return result
stock = StockTool()
stock.invoke({'ticker': 'AAPL'}) # → 'AAPL: $100.5'
stock.invoke({'ticker': 'GOOGL', 'include_history': True}) # → 'GOOGL: $102.1 (30일 변동: +5.11)'
stock.query_log # → [{'ticker': 'AAPL', 'time': ...}, ...]
|
오케스트레이터는 실행 순서와 결과 결합에 집중시키는 편이 읽기 쉽다. 개별 도구의 요청·검증을 한곳에 계속 추가하기보다 공통 반환 형태를 맞추고, 상위 단계에서 어떤 실패를 허용할지 결정한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
| @tool
def api_get_user(user_id):
"""사용자 기본 정보(이름, 이메일, 도시)를 조회합니다"""
r = requests.get(f"{BASE_URL}/users/{user_id}")
u = r.json()
return json.dumps(
{'name': u['name'], 'email': u['email'], 'city': u['address']['city']},
ensure_ascii=False
)
@tool
def api_get_user_posts(user_id, limit=5):
"""사용자의 최근 게시글 제목을 조회합니다"""
r = requests.get(f"{BASE_URL}/posts", params={'userId': user_id})
posts = r.json()[:limit]
return json.dumps([{'id': p['id'], 'title': p['title']} for p in posts], ensure_ascii=False)
@tool
def api_get_user_todos(user_id):
"""사용자의 TODO 완료율을 조회합니다"""
r = requests.get(f"{BASE_URL}/todos", params={'userId': user_id})
todos = r.json()
done = sum(1 for t in todos if t['completed'])
return json.dumps(
{'completed': done, 'total': len(todos), 'rate': f'{done/len(todos)*100:.0f}%'},
ensure_ascii=False
)
api_tools = [api_get_user, api_get_user_posts, api_get_user_todos]
tool_map = {t.name: t for t in api_tools}
def orchestrate(query, max_turns=5):
llm_api = llm.bind_tools(api_tools)
messages = [HumanMessage(content=query)]
tool_call_count = 0
for turn in range(max_turns):
try:
resp = llm_api.invoke(messages)
except Exception as e:
return f"llm call fail: {e}"
messages.append(resp)
if not resp.tool_calls:
return resp.content
for tc in resp.tool_calls:
name = tc['name']
if name not in tool_map:
result = f"Error: unknown tool '{name}'"
else:
tool_call_count += 1
if tool_call_count > 10:
return "Error: tool call limit exceeded"
try:
result = tool_map[tc['name']].invoke(tc['args'])
except:
result = "Error"
print(f" [Turn {turn+1}] {name} ({tc['args']}) → {str(result)[:80]}")
messages.append(ToolMessage(content=result, tool_call_id=tc['id']))
return '최대 턴 초과'
orchestrate('1번 유저의 정보와 최근 게시글 3개, todo 완료율을 알려줘')
|
11. 핵심 개념 총정리
도구 패턴 선택 기준
1
2
3
4
| 단순 함수 → @tool
복잡한 입력 검증 필요 → StructuredTool.from_function + BaseModel
캐시/카운터/상태 필요 → BaseTool 상속
캐시 + Pydantic 스키마 → BaseTool + args_schema
|
HTTP 도구 만들기 체크리스트
1
2
3
4
5
| 1. r.ok 로 에러 확인
2. return f"error: {r.status_code}" 로 에러 메시지 반환
3. json.dumps(... , ensure_ascii=False) 로 한글 인코딩
4. 결과를 str로 변환해서 ToolMessage.content에 전달
5. timeout 설정
|
12. 자주 나오는 실수/주의사항
실수 1: ensure_ascii=False 빠뜨리기
1
2
3
4
5
6
7
| # 틀린 코드 → 한글이 \uXXXX 형태로 인코딩됨
return json.dumps({'name': '홍길동'})
# → '{"name": "\\ud64d\\uae38\\ub3d9"}'
# 올바른 코드
return json.dumps({'name': '홍길동'}, ensure_ascii=False)
# → '{"name": "홍길동"}'
|
실수 2: ToolMessage.content에 dict 그대로 전달
1
2
3
4
5
6
7
| # 틀린 코드 → content는 str이어야 함
messages.append(ToolMessage(content={'name': '홍길동'}, tool_call_id=tc['id']))
# 올바른 코드
messages.append(ToolMessage(content=json.dumps({'name': '홍길동'}, ensure_ascii=False), tool_call_id=tc['id']))
# 또는
messages.append(ToolMessage(content=str(result), tool_call_id=tc['id']))
|
1
2
3
4
5
6
7
| # 틀린 코드 → Pydantic 모델에서 필드 선언 없으면 오류
class MyTool(BaseTool):
cache = {} # 타입 없음 → 에러
# 올바른 코드
class MyTool(BaseTool):
cache: dict = {} # 타입 선언 필수
|
실수 4: 페이지 번호 없이 params에 userId와 _page 혼용
1
2
3
4
| # /todos에는 _page 파라미터가 작동함
r = requests.get(f"{BASE_URL}/todos", params={'userID': user_id})
# 주의: userId (소문자 i) 가 올바른 파라미터명
# userID (대문자 D) 는 무시될 수 있음 → 전체 목록 반환됨
|
13. 부분 성공을 결과에 표현하기
리포트 생성 성공을 파일이 만들어졌는지로만 판단하지 않는다. 필수 데이터가 모였는지와 선택 항목이 빠졌는지를 구분하고, 사용자에게 전달하는 요약에서도 그 범위를 유지한다.
확인할 사례는 모든 API 성공, 한 API 실패, 빈 응답, 필드 누락이다. 이 경우들을 같은 반환 구조로 다룰 수 있다면 도구를 추가할 때도 리포트 코드의 분기가 과도하게 늘어나지 않는다.