第 11 章 · ReAct Agent 实现 & create_react_agent 复盘#

1
mkdir -p scripts/ch11
2
touch scripts/ch11/test_react_agent.py

1
import asyncio
2

3
from langchain_core.messages import HumanMessage
4
from langgraph.prebuilt import create_react_agent
5

6
from app.ai.llm import llm
7
from app.ai.tools import get_weather, calculate, get_current_time
8

9
# 工具列表：Agent 能自主决定调用其中的哪几个
10
tools = [get_weather, calculate, get_current_time]
11

12
# 一行代码创建完整的 ReAct Agent
13
# create_react_agent 内部已经帮我们搭好了 LangGraph 的图结构
14
agent = create_react_agent(llm, tools)
15

16
# 打印 Agent 内部的 Graph 结构（Mermaid 文本），直观看到它搭了什么
17
print("ReAct Agent 内部 Graph 结构：")
18
print(agent.get_graph().draw_mermaid())
19

20

21
async def run_agent(question: str):
22
    print(f"\n{'=' * 50}")
23
    print(f"用户问题：{question}")
24
    print("-" * 50)
25

26
    result = await agent.ainvoke({
27
        "messages": [HumanMessage(content=question)]
28
    })
29

30
    # 遍历所有消息，复盘 Agent 的完整思考过程
31
    for msg in result["messages"]:
32
        msg_type = type(msg).__name__
33

34
        if hasattr(msg, "tool_calls") and msg.tool_calls:
35
            print(f"[{msg_type}] 决定调用工具：{[tc['name'] for tc in msg.tool_calls]}")
36
            for tc in msg.tool_calls:
37
                print(f"  工具名：{tc['name']}，参数：{tc['args']}")
38
        elif msg_type == "ToolMessage":
39
            print(f"[{msg_type}] 工具返回：{msg.content}")
40
        elif msg_type == "AIMessage" and msg.content:
41
            print(f"[{msg_type}] AI 回复：{msg.content}")
42

43
    print(f"\n最终答案：{result['messages'][-1].content}")
44

45

46
async def main():
47
    # 场景一：单工具调用（实时天气）
48
    await run_agent("武汉现在天气怎么样？")
49
    # 场景二：数学计算（走 calculate 工具，而不是让模型心算）
50
    await run_agent("帮我算一下 (256 + 144) / 20 等于多少")
51
    # 场景三：不需要工具，模型直接回答
52
    await run_agent("Python 的 list 和 tuple 有什么区别？")
53

54

55
asyncio.run(main())

1
uv run python scripts/ch11/test_react_agent.py

1
import asyncio
2
from typing import Annotated
3
from typing_extensions import TypedDict
4

5
from langchain_core.messages import HumanMessage
6
from langgraph.graph import StateGraph, START, END
7
from langgraph.graph.message import add_messages
8
from langgraph.prebuilt import ToolNode, tools_condition
9

10
from app.ai.llm import llm
11
from app.ai.tools import get_weather, calculate, get_current_time
12

13
tools = [get_weather, calculate, get_current_time]
14

15

16
# 1. 定义状态：messages 借助 add_messages 自动累加
17
class ReactState(TypedDict):
18
    messages: Annotated[list, add_messages]
19

20

21
# 2. chat 节点：调用绑定工具后的 LLM，产出 AIMessage（可能带 tool_calls）
22
def chat_node(state: ReactState):
23
    llm_with_tools = llm.bind_tools(tools)
24
    ai_message = llm_with_tools.invoke(state["messages"])
25
    return {"messages": [ai_message]}
26

27

28
# 3. tools 节点：ToolNode 自动读取上一条 AIMessage 里的 tool_calls 并执行
29
tool_node = ToolNode(tools)
30

31

32
# 4. 搭图
33
graph_builder = StateGraph(ReactState)
34
graph_builder.add_node("chat", chat_node)
35
graph_builder.add_node("tools", tool_node)
36

37
# 入口 -> chat
38
graph_builder.add_edge(START, "chat")
39

40
# chat 之后条件路由：
41
#   AIMessage 带 tool_calls -> 去 "tools"
42
#   AIMessage 不带 tool_calls -> END（输出最终答案）
43
graph_builder.add_conditional_edges("chat", tools_condition)
44

45
# 工具执行完 -> 回到 chat，让模型基于 ToolMessage 继续推理
46
graph_builder.add_edge("tools", "chat")
47

48
# 编译得到可运行的 graph
49
manual_agent = graph_builder.compile()
50

51

52
# 5.（可选）不使用 tools_condition，手写条件路由，逻辑完全等价
53
def should_continue(state: ReactState) -> str:
54
    last_message = state["messages"][-1]
55
    if getattr(last_message, "tool_calls", None):
56
        return "tools"   # 模型请求调工具
57
    return END           # 模型给出最终回答，结束循环
58

59

60
# 用自定义路由替换内置条件边：
61
# graph_builder.add_conditional_edges("chat", should_continue, {"tools": "tools", END: END})
62

63

64
async def main():
65
    # 打印手写图结构，和 create_react_agent 打印的结构对照
66
    print(manual_agent.get_graph().draw_mermaid())
67

68
    result = await manual_agent.ainvoke({
69
        "messages": [HumanMessage(content="帮我算一下 (256 + 144) / 20，再告诉我现在几点")]
70
    })
71
    for msg in result["messages"]:
72
        print(f"[{type(msg).__name__}] {msg.content if msg.content else msg.tool_calls}")
73

74

75
asyncio.run(main())

1
from langchain_core.messages import HumanMessage
2
from langgraph.prebuilt import create_react_agent
3

4
from app.ai.llm import llm
5
from app.ai.tools import get_weather, calculate, get_current_time
6

7
# 全局只构建一次 Agent（编译图有开销，避免每次请求重建）
8
tools = [get_weather, calculate, get_current_time]
9

10
SYSTEM_PROMPT = (
11
    "你是一个智能助手，优先使用工具获取实时信息和完成精确计算，"
12
    "使用中文简洁回复。"
13
)
14

15
react_agent = create_react_agent(
16
    llm,
17
    tools,
18
    state_modifier=SYSTEM_PROMPT,
19
)
20

21

22
class AgentService:
23
    """对外暴露的 Agent 服务：普通调用 + 流式调用"""
24

25
    @staticmethod
26
    async def chat(question: str) -> str:
27
        """一次性返回最终答案"""
28
        result = await react_agent.ainvoke({
29
            "messages": [HumanMessage(content=question)]
30
        })
31
        return result["messages"][-1].content
32

33
    @staticmethod
34
    async def stream_chat(question: str):
35
        """异步生成器：只把模型的文本 token 逐块吐出，工具过程可按需透传"""
36
        async for event in react_agent.astream_events(
37
            {"messages": [HumanMessage(content=question)]},
38
            version="v2",
39
        ):
40
            if event["event"] == "on_chat_model_stream":
41
                chunk = event["data"].get("chunk")
42
                if chunk and getattr(chunk, "content", None):
43
                    yield chunk.content

1
from fastapi import APIRouter
2
from fastapi.responses import StreamingResponse
3
from pydantic import BaseModel
4

5
from app.ai.graphs.react_agent import AgentService
6

7
router = APIRouter(prefix="/chat", tags=["chat"])
8

9

10
class ChatRequest(BaseModel):
11
    question: str
12

13

14
@router.post("/agent")
15
async def agent_chat(req: ChatRequest):
16
    """普通接口：等 Agent 跑完整个 ReAct 循环后一次性返回"""
17
    answer = await AgentService.chat(req.question)
18
    return {"question": req.question, "answer": answer}
19

20

21
@router.post("/agent/stream")
22
async def agent_stream(req: ChatRequest):
23
    """SSE 流式接口：边推理边返回 token"""
24

25
    async def event_generator():
26
        async for token in AgentService.stream_chat(req.question):
27
            # SSE 协议：每条消息以 data: 开头，两个换行结束
28
            yield f"data: {token}\n\n"
29
        yield "data: [DONE]\n\n"
30

31
    return StreamingResponse(
32
        event_generator(),
33
        media_type="text/event-stream",
34
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
35
    )

1
from fastapi import FastAPI
2
from app.routers import chat
3

4
app = FastAPI(title="AI Agent Demo")
5
app.include_router(chat.router)

1
curl -N -X POST "http://localhost:8000/chat/agent/stream" \
2
  -H "Content-Type: application/json" \
3
  -d '{"question": "现在几点了？(128+72)*3 等于多少？"}'