一、程式碼範例(Python)
from typing import TypedDict
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, ToolMessage
from langchain_core.tools import tool
from pydantic import BaseModel, Field
import os
import requests
# 定義 state 的結構
class MyAgentState(TypedDict):
messages: list[BaseMessage] # 用來儲存對話訊息
counter: int
# 定義工具(tool)
class WeatherInput(BaseModel):
location: str = Field(..., description="城市名稱,例如 Taipei")
date: str = Field(..., description="查詢日期,格式 YYYY-MM-DD")
@tool("get_weather", args_schema=WeatherInput, return_direct=True)
def get_weather(location: str, date: str):
# 簡單示意,實際要換成真的天氣 API
response = requests.get(
f"https://api.example.com/weather?loc={location}&date={date}"
)
if response.status_code == 200:
return {"weather": response.json()}
else:
return {"error": f"Status code {response.status_code}"}
# 定義 node 函式
def call_model_node(state: MyAgentState):
"""
用 LLM 處理最新對話,看是否要呼叫工具
假設 state["messages"] 最後一條訊息可能包含 tool_calls
"""
# 這裡假設已有一個 model_with_tools
# model_with_tools.invoke(...) 回傳 BaseMessage
# 為了示意,我這裡直接回傳一個 ToolMessage 或普通訊息
# 假設 last message 想查天氣
tool_msg = ToolMessage(
content=None,
name="get_weather",
tool_call_id="call1",
# args 跟 tool schema 相符
**{"args": {"location": "Taipei", "date": "2025-09-21"}}
)
return {"messages": [tool_msg]}
def call_tool_node(state: MyAgentState):
"""
處理 tool 呼叫
"""
outputs = []
last_msg = state["messages"][-1]
if isinstance(last_msg, ToolMessage):
# 執行工具
res = get_weather.invoke(**last_msg.tool_call_args) # 假設有這屬性
outputs.append(
ToolMessage(
content=res,
name=last_msg.name,
tool_call_id=last_msg.tool_call_id,
)
)
return {"messages": outputs, "counter": state["counter"] + 1}
def decide_next(state: MyAgentState):
"""決定流程是否結束或繼續呼叫 model 或 tool"""
last_msg = state["messages"][-1]
# 如果最後訊息不是 ToolMessage,就結束
if not isinstance(last_msg, ToolMessage):
return END
# 否則回到 model node 繼續處理
return "call_model"
# 建構 graph
workflow = StateGraph(state_schema=MyAgentState)
workflow.add_node("call_model", call_model_node)
workflow.add_node("call_tool", call_tool_node)
workflow.set_entry_point("call_model")
workflow.add_edge("call_model", "call_tool")
workflow.add_conditional_edges(source="call_tool", path=decide_next)
# 初始化 state
initial_state: MyAgentState = {
"messages": [],
"counter": 0,
}
# 執行 graph
graph = workflow.compile()
final_state = graph.invoke(initial_state)
print(final_state)
二、說明
名稱 |
用途 |
StateGraph |
定義整個工作流程的圖(graph),包含狀態 (state)、節點 (nodes)、邊 (edges) |
state_schema |
定義狀態裡會有哪些欄位 (fields),也規範類型 |
@tool |
定義一個工具(tool),可以在流程中被節點呼叫 |
node 函式 |
每個 node 接收目前的 state,回傳該 node 處理後的部分 state 變更 |
edge |
普通邊 (無條件) 和條件邊 (conditional edge);條件邊可以依 state 決定接下來要去哪個 node 或是結束流程 |
END / START |
LangGraph 提供的特殊標記,用來指示流程結束或開始的地方 |
三、資料來源