15. 使用 LangGraph 构建你的第一个图

# 第 14 章代理循环 — 核心结构(简化版)
messages = [
    SystemMessage(content="你是一个有帮助的助手。"),
    HumanMessage(content=user_input),
]
 
for step in range(max_steps):
    # 请求 LLM 决定下一步操作
    ai_message = llm_with_tools.invoke(messages)
    messages.append(ai_message)
 
    # 如果没有请求工具调用,返回最终答案
    if not ai_message.tool_calls:
        return ai_message.content
 
    # 执行请求的工具
    for tool_call in ai_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        messages.append(tool_message)

from langgraph.graph import StateGraph, MessagesState
 
builder = StateGraph(MessagesState)

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder.add_node(say_hello)    # 节点名变为 "say_hello"

from langgraph.graph import START, END
 
builder.add_edge(START, "say_hello")   # 图开始 → 运行 say_hello
builder.add_edge("say_hello", END)     # say_hello 完成 → 结束图

graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)

from langgraph.graph import StateGraph, MessagesState, START, END
 
def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder = StateGraph(MessagesState)
 
builder.add_node(say_hello)
builder.add_edge(START, "say_hello")
builder.add_edge("say_hello", END)
 
graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)
print(result["messages"][-1].content)

hello world

from typing_extensions import TypedDict
 
class AgentState(TypedDict):
    messages: list       # 消息列表
    llm_calls: int       # LLM 调用次数

from typing_extensions import TypedDict, Annotated
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # 归约器:追加
    llm_calls: int                            # 无归约器:覆盖

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int  # 默认覆盖行为

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

builder.add_node(say_hello)          # 函数名 "say_hello" 变为节点名
builder.add_node("my_node", my_func) # 你也可以显式指定名称

builder.add_edge(START, "say_hello")       # 图开始时,运行 say_hello
builder.add_edge("say_hello", "llm_call")  # say_hello 之后,运行 llm_call
builder.add_edge("llm_call", END)          # llm_call 之后,结束图

from langgraph.graph import END
 
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"   # 请求了工具调用 → 前往 tool_node
    return END               # 没有工具调用 → 结束
 
builder.add_conditional_edges("llm_call", should_continue)

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int    # LLM 调用次数(覆盖)

from langchain_openai import ChatOpenAI
from langchain_core.messages import ToolMessage
from langchain.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """获取某个城市的当前天气。"""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """计算一个简单的算术表达式。示例: '3 * 21'。"""
    return str(eval(expression))  # 警告:eval() 存在安全风险。请勿在生产环境中使用。
 
tools = [get_weather, calculate]
tool_map = {t.name: t for t in tools}
 
llm = ChatOpenAI(model="gpt-5-mini")
model_with_tools = llm.bind_tools(tools)

def llm_call(state: AgentState):
    """调用 LLM 并返回响应。"""
    response = model_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

def tool_node(state: AgentState):
    """执行 LLM 请求的工具。"""
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        results.append(tool_message)
    return {"messages": results}

from typing import Literal
from langgraph.graph import END
 
def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
    """决定是执行工具还是结束图。"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(AgentState)
 
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
 
builder.add_edge(START, "llm_call")                         # 开始 → llm_call
builder.add_conditional_edges("llm_call", should_continue)  # llm_call → tool_node 或 END
builder.add_edge("tool_node", "llm_call")                   # tool_node → llm_call(循环)
 
agent = builder.compile()

from langchain_core.messages import HumanMessage
 
result = agent.invoke({
    "messages": [HumanMessage(content="获取开罗的气温,然后将这个数字乘以 3。")],
    "llm_calls": 0,
})
 
print(result["messages"][-1].content)
print(f"\nTotal LLM calls: {result['llm_calls']}")

开罗当前气温:31°C。乘以 3 = 93。
 
Total LLM calls: 3

for message in result["messages"]:
    message.pretty_print()

================================ Human Message =================================
获取开罗的气温,然后将这个数字乘以 3。
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_DiL9WF)
  Args:
    city: Cairo
================================= Tool Message =================================
Name: get_weather
31°C, sunny
================================== Ai Message ==================================
Tool Calls:
  calculate (call_wa6RqWST)
  Args:
    expression: 31 * 3
================================= Tool Message =================================
Name: calculate
93
================================== Ai Message ==================================
开罗当前气温:31°C。乘以 3 = 93。

from IPython.display import Image, display
 
display(Image(agent.get_graph().draw_mermaid_png()))

agent.get_graph().draw_mermaid_png(output_file_path="agent_graph.png")

from langgraph.errors import GraphRecursionError
 
try:
    result = agent.invoke(
        {"messages": [HumanMessage(content="获取开罗的气温,然后将这个数字乘以 3。")],
         "llm_calls": 0},
        config={"recursion_limit": 3},
    )
except GraphRecursionError:
    print("代理达到了递归限制 — 停止执行。")

代理达到了递归限制 — 停止执行。