TIL

Chapter 4 도구

4.1 랭체인 기초

from langchain.chat_models import init_chat_model
llm = init_chat_model(model="gpt-5-mini", temperature=0)

from langchain_core.messages import HumanMessage
messages = [HumanMessage("오늘 날씨는?")]

from langchain_core.tools import tool

@tool
def add(x: float, y: float) -> float: 
	"""'x'와 'y'를 더합니다."""
	return x + y

# 도구 목록을 LLM에 바인딩 → LLM이 응답 시 이 도구를 선택할 수 있게 됨
llm_with_tools = llm.bind_tools([add])

# 메시지를 보내면 LLM이 직접 답하거나, 도구 호출(tool_call)을 반환
ai_msg = llm_with_tools.invoke(messages)

# LLM이 요청한 도구 호출을 순회하며 실제 함수 실행
for tool_call in ai_msg.tool_calls:
	tool_response = add.invoke(tool_call) # 함수 실행 결과를 ToolMessage 형태로 반환

4.1.1 로컬 도구

# 1. 도구 정의
@tool
def multiply(x: float, y: float) -> float:
    """두 수를 곱합니다."""
    return x * y

@tool
def divide(x: float, y: float) -> float:
    """x를 y로 나눕니다."""
    return x / y

# 2. 바인딩
llm_with_tools = llm.bind_tools([multiply, divide])

# 3. 호출 — LLM이 적절한 도구를 선택해 tool_call 반환
messages = [HumanMessage(content="120 나누기 7은?")]
ai_msg = llm_with_tools.invoke(messages)
messages.append(ai_msg)

# 4. 도구 실행 — 이름으로 매핑해서 LLM이 선택한 도구를 찾아 실행
tools = {"multiply": multiply, "divide": divide}

for tool_call in ai_msg.tool_calls:
    tool = tools[tool_call["name"]]
    messages.append(tool.invoke(tool_call))

# 5. 재호출 — 도구 결과를 보고 LLM이 최종 응답 생성
final = llm_with_tools.invoke(messages) # "120 나누기 7은 약 17.14입니다."

4.1.2 API 기반 도구

4.1.3 플러그인 도구

4.1.4 MCP

"""
LangGraph × MCP 클라이언트 학습 예제
──────────────────────────────────
- math    : 로컬 stdio 서브프로세스로 실행되는 MCP 서버
- weather : 원격 streamable_http MCP 서버 (미리 기동해두어야 함)

그래프는 노드가 단 하나뿐인 '미니 에이전트'.
실제 에이전트라면 LLM 이 도구를 고르겠지만, 여기서는 규칙 기반으로
라우팅해 MCP 호출 흐름 자체에 집중한다.
"""
import asyncio
import re
from typing import Any, Sequence, TypedDict

from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph

class AgentState(TypedDict):
    messages: Sequence[Any]

# ── 1) 두 MCP 서버를 한 클라이언트에 등록 ────────────────────────
mcp_client = MultiServerMCPClient({
    "math": {
        "command": "python3",
        "args": ["ch04/mcp_servers/MCP_math_server.py"],
        "transport": "stdio",                  # 서브프로세스 + stdin/stdout 파이프
    },
    "weather": {
        # 먼저 `python ch04/mcp_servers/MCP_weather_server.py` 로 기동
        "url": "http://0.0.0.0:8000/mcp",
        "transport": "streamable_http",        # 원격 HTTP 스트리밍
    },
})

# get_tools() 는 서버에 질의하는 비용이 있으므로 최초 1회만 호출해 캐시
_tools: list | None = None
async def get_tools() -> list:
    global _tools
    if _tools is None:
        _tools = await mcp_client.get_tools()
    return _tools

# ── 2) 규칙 기반 라우팅: 마지막 사용자 메시지 → (도구명, 입력) ──
_HAS_OP = re.compile(r"[+\-*/()]")             # 연산자 하나라도 있으면 수식 취급
_WEATHER_IN = re.compile(r"weather in\s+(.+?)\??$", re.I)
_UI_NALSSI  = re.compile(r"(.+?)\s*의\s*날씨")

def route(text: str) -> tuple[str, dict] | None:
    if _HAS_OP.search(text):
        return "math", {"expression": text}
    if "weather" in text.lower() or "날씨" in text:
        m = _WEATHER_IN.search(text) or _UI_NALSSI.search(text)
        return "weather", {"location": m.group(1).strip() if m else "NYC"}
    return None

# ── 3) 단일 그래프 노드: 라우팅 결과대로 MCP 도구를 호출 ─────────
async def call_mcp_tools(state: AgentState) -> dict[str, Any]:
    user_text = state["messages"][-1].content
    picked = route(user_text)

    if picked is None:
        reply = "수학 또는 날씨 질문만 답변할 수 있습니다."
    else:
        name, tool_input = picked
        tool = next((t for t in await get_tools() if t.name == name), None)
        reply = await tool.ainvoke(tool_input) if tool else f"{name} 도구를 사용할 수 없습니다."

    return {"messages": [{"role": "assistant", "content": reply}]}

# ── 4) 노드 1개짜리 StateGraph 컴파일 ───────────────────────────
def build_graph():
    g = StateGraph(AgentState)
    g.add_node("assistant", call_mcp_tools)
    g.set_entry_point("assistant")
    return g.compile()

GRAPH = build_graph()

# ── 5) 실행 헬퍼 ────────────────────────────────────────────────
async def ask(question: str) -> str:
    """질문 한 건을 그래프에 넣고 assistant 의 최종 응답 텍스트만 뽑는다."""
    result = await GRAPH.ainvoke({"messages": [HumanMessage(content=question)]})
    msg = result["messages"][-1]
    return msg["content"] if isinstance(msg, dict) else msg.content

async def main():
    print("Math    ▶", await ask("(3 + 5) * 12은 얼마인가요?"))
    print("Weather ▶", await ask("NYC의 날씨는 어때요?"))

if __name__ == "__main__":
    asyncio.run(main())

4.1.5 상태 유지 도구

4.2 도구 개발 자동화

4.2.1 파운데이션 모델을 활용한 도구 개발

4.2.2 실시간 코드 생성

4.3 도구 사용 설정