跳到主要内容

Agent 架构设计

构建一个高效、可靠的 AI Agent 需要深入理解其架构设计。本章将介绍 Agent 的核心架构模式、设计原则以及常见的设计决策。

LangGraph Agent 架构​

LangGraph 提供了基于图结构的 Agent 运行时。这种架构提供了更好的可控性、可观测性和可扩展性。

核心概念：Graph-based Runtime​

LangGraph 的 Agent 运行时基于图（Graph）结构构建。图由节点（Nodes）和边（Edges）组成：

  • 节点：代表处理步骤，如模型调用节点、工具执行节点
  • 边：定义节点之间的流转关系，支持条件分支

Agent 在图中移动，执行各个节点，直到达到停止条件（模型输出最终答案或达到迭代限制）。

create_agent 和 create_react_agent API​

LangChain 生态系统提供两种创建 Agent 的方式：

方式一：create_agent（支持 middleware）​

LangChain 1.0 新增的 API，原生支持 middleware 机制：

from langchain.agents import create_agent
from langchain.tools import tool

# 定义工具
@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果：{query}"

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except Exception:
        return "计算错误"

# 创建 Agent
# 方式一：使用模型字符串（最简单）
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[search, calculator],
    system_prompt="你是一个智能助手，可以帮助用户搜索信息和进行计算。"
)

# 方式二：使用模型实例
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_agent(
    model=llm,
    tools=[search, calculator],
    system_prompt="你是一个智能助手，可以帮助用户搜索信息和进行计算。"
)

# 调用 Agent
result = agent.invoke({"messages": [{"role": "user", "content": "帮我搜索 Python 教程"}]})
print(result["messages"][-1].content)

方式二：create_react_agent（简洁易用）​

LangGraph 提供的经典 API：

from langgraph.prebuilt import create_react_agent
from langchain.tools import tool

@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果：{query}"

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except Exception:
        return "计算错误"

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[search, calculator],
    prompt="你是一个智能助手。"  # 注意：参数名是 prompt
)

API 对比​

特性create_agentcreate_react_agent
导入路径langchain.agentslanggraph.prebuilt
系统提示词参数system_promptprompt
中间件支持支持 middleware 参数不支持
自定义状态仅支持 TypedDict支持 Pydantic/Dataclass/TypedDict
流式输出if "agent" in chunkif "agent" in chunk

create_agent 参数详解​

create_agent 函数支持以下核心参数：

参数类型说明
modelstr 或 BaseChatModel语言模型，可以是字符串（如 "openai:gpt-4o-mini"）或模型实例
toolsSequence[BaseTool]工具列表
system_promptstr / SystemMessage系统提示词，定义 Agent 行为
checkpointerCheckpointer短期记忆持久化，用于保存对话状态
storeBaseStore长期记忆存储，跨会话保存数据
response_formatSchema结构化输出格式定义（使用新的策略）
state_schemaTypedDict自定义状态结构（仅支持 TypedDict）
middlewarelist[Middleware]中间件列表，用于扩展 Agent 能力
interrupt_beforelist[str]在指定节点前中断执行
interrupt_afterlist[str]在指定节点后中断执行
debugbool是否启用调试模式

流式输出​

使用 stream 方法可以看到 Agent 的完整执行过程：

# 流式输出（可以看到 Agent 的思考过程）
for chunk in agent.stream({
    "messages": [{"role": "user", "content": "计算 100 / 4 + 20"}]
}):
    # 注意：节点名从 "agent" 改为 "model"
    if "model" in chunk:
        for msg in chunk["model"]["messages"]:
            print(f"[Model] {msg.content}")
    elif "tools" in chunk:
        for msg in chunk["tools"]["messages"]:
            print(f"[Tool] {msg.content}")

核心组件详解​

Model（模型）​

模型是 Agent 的推理引擎，负责理解用户意图和生成响应。LangChain 1.0 支持多种方式指定模型：

使用模型字符串（最简单的方式）：

from langchain.agents import create_agent

# 直接使用模型字符串
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    system_prompt="你是一个智能助手。"
)

使用模型实例（支持更多配置）：

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

# 创建配置化的模型实例
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,      # 降低随机性
    max_tokens=2000,    # 限制输出长度
    timeout=30          # 设置超时
)

agent = create_agent(model=llm, tools=[...])

动态选择模型（使用 middleware）：

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain.chat_models import init_chat_model

class DynamicModelMiddleware(AgentMiddleware):
    """动态模型选择中间件"""
    
    def __init__(self, use_powerful_model: bool = False):
        self.use_powerful_model = use_powerful_model
        self.default_model = init_chat_model("openai:gpt-4o-mini")
        self.powerful_model = init_chat_model("openai:gpt-4o")
    
    def wrap_model_call(self, request, handler):
        """在模型调用前选择模型"""
        if self.use_powerful_model:
            modified_request = request.copy()
            modified_request["model"] = self.powerful_model
            return (yield handler(modified_request))
        return (yield handler(request))

agent = create_agent(
    model="openai:gpt-4o-mini",  # 默认模型
    tools=[...],
    middleware=[DynamicModelMiddleware(use_powerful_model=True)]
)

Tools（工具）​

工具赋予 Agent 与外部世界交互的能力。LangChain 1.0 提供了强大的工具管理能力：

  • 多工具顺序调用
  • 工具并行执行
  • 动态工具选择
  • 工具错误处理（通过 middleware）

定义工具：

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取城市天气信息
    
    参数：
    - city: 城市名称，如：北京、上海
    """
    weather_data = {
        "北京": {"temp": 22, "condition": "晴"},
        "上海": {"temp": 26, "condition": "多云"},
    }
    data = weather_data.get(city, {"temp": 20, "condition": "未知"})
    return f"{city}：{data['condition']}，温度 {data['temp']}°C"

@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果：{query}"

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_weather, search]
)

使用 InjectedState 访问状态：

在 LangChain 1.0 中，推荐使用 middleware 来管理状态访问。工具可以通过 context 参数获取运行时上下文：

from langchain.tools import tool
from langchain.agents import AgentState
from typing import TypedDict

class CustomState(AgentState):
    user_id: str

@tool
def get_user_data(user_id: str) -> str:
    """获取当前用户数据"""
    return f"用户 {user_id} 的数据"

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_user_data],
    state_schema=CustomState
)

System Prompt（系统提示词）​

系统提示词定义了 Agent 的行为方式和任务导向。在 LangChain 1.0 中，参数名从 prompt 改为 system_prompt：

from langchain.agents import create_agent

# 方式一：字符串形式
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    system_prompt="你是一个专业的客服助手，请用礼貌、专业的语气回答用户问题。"
)

# 方式二：使用 SystemMessage
from langchain.messages import SystemMessage

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    system_prompt=SystemMessage(
        content="你是一个智能助手，具备以下能力：\n"
                "1. 信息搜索和分析\n"
                "2. 数学计算\n"
                "3. 任务规划和执行"
    )
)

动态提示词（使用 @dynamic_prompt 装饰器）：

from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt
from datetime import datetime

@dynamic_prompt
def create_system_prompt(state, context):
    """动态生成系统提示词"""
    today = datetime.now().strftime("%Y-%m-%d")
    return f"你是一个智能助手。当前日期：{today}"

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    system_prompt=create_system_prompt
)

State（状态）​

状态在 Agent 执行过程中自动维护，包含对话历史和自定义数据。在 LangChain 1.0 中，自定义状态只支持 TypedDict：

from typing import TypedDict
from langchain.agents import create_agent, AgentState

class MyState(AgentState):
    """扩展 Agent 状态（仅支持 TypedDict）"""
    user_id: str
    preferences: dict
    # messages 字段已在 AgentState 中定义

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    state_schema=MyState
)

# 调用时传入自定义状态
result = agent.invoke({
    "messages": [{"role": "user", "content": "你好"}],
    "user_id": "user-123",
    "preferences": {"language": "zh"}
})

注意：LangChain 1.0 不再支持使用 Pydantic BaseModel 或 dataclass 作为状态 schema。如果需要验证，请在 middleware 中处理。

Checkpointer 和 Store​

LangGraph 使用 checkpointer 管理短期记忆，使用 store 管理长期记忆：

短期记忆（Checkpointer）：

from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver

# 创建 checkpointer（内存存储，适合开发测试）
checkpointer = MemorySaver()

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    checkpointer=checkpointer
)

# 使用 thread_id 区分会话
config = {"configurable": {"thread_id": "session-001"}}

# 第一次对话
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫张三"}]},
    config=config
)

# 第二次对话（Agent 会记住之前的内容）
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config=config
)
print(result2["messages"][-1].content)  # 输出：您叫张三。

长期记忆（Store）：

from langchain.agents import create_agent
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool
import uuid

# 创建 store
store = InMemoryStore()

@tool
def save_memory(content: str) -> str:
    """保存信息到长期记忆"""
    return f"已保存：{content}"

@tool
def recall_memories(query: str) -> str:
    """从长期记忆中检索信息"""
    return "记忆内容..."

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[save_memory, recall_memories],
    store=store
)

# 保存记忆
agent.invoke(
    {"messages": [{"role": "user", "content": "记住我喜欢喝咖啡"}]},
    config={"configurable": {"user_id": "user-123"}}
)

Middleware 和错误处理​

LangChain 1.0 使用 middleware 机制来处理工具错误，替代了之前的 ToolNode 配置：

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain.tools import tool

@tool
def risky_operation(param: str) -> str:
    """可能失败的操作"""
    if param == "error":
        raise ValueError("操作失败")
    return f"成功：{param}"

class ErrorHandlingMiddleware(AgentMiddleware):
    """错误处理中间件"""
    
    def wrap_tool_call(self, request, handler):
        """包装工具调用，处理错误"""
        try:
            result = yield handler(request)
            return result
        except Exception as e:
            # 返回友好的错误消息
            return f"操作执行失败：{str(e)}。请尝试其他方法。"

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[risky_operation],
    middleware=[ErrorHandlingMiddleware()]
)

自定义错误消息：

from langchain.agents.middleware import AgentMiddleware

class CustomErrorMiddleware(AgentMiddleware):
    """自定义错误处理中间件"""
    
    def wrap_tool_call(self, request, handler):
        """包装工具调用，处理错误"""
        try:
            result = yield handler(request)
            return result
        except ValueError as e:
            return f"参数错误：{str(e)}"
        except ConnectionError as e:
            return "网络连接失败，请稍后重试"
        except Exception as e:
            return f"未知错误：{str(e)}"

核心架构模式​

单 Agent 架构​

最简单的 Agent 架构由一个 LLM 和一组工具组成：

这种架构适合处理单一领域的任务，实现简单，易于调试。但当任务变得复杂时，单个 Agent 可能难以有效管理。

适用场景：

  • 功能明确的专业助手（如天气查询、翻译）
  • 流程相对固定的自动化任务
  • 原型开发和概念验证

代码示例：

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果：{query}"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except Exception:
        return "计算错误"

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[search, calculate],
    system_prompt="你是一个智能助手，可以搜索信息和进行计算。"
)

# 简单调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "搜索 Python 教程并计算 2 + 2"}]
})

多 Agent 协作架构​

对于复杂任务，可以使用多个 Agent 协作完成：

每个 Agent 专注于特定领域，通过共享记忆或消息传递进行协作。

适用场景：

  • 需要多领域知识的复杂任务
  • 大规模自动化流程
  • 需要分工合作的场景

层级 Agent 架构​

层级架构模拟人类组织结构，由管理者 Agent 分配任务给执行者 Agent：

监督者负责任务分配和结果整合，规划者负责制定执行计划，执行者负责具体实施。

ReAct 模式​

ReAct（Reasoning and Acting）是目前最流行的 Agent 执行模式，它将推理和行动紧密结合。LangGraph 的 create_react_agent 默认采用 ReAct 模式。

ReAct 的工作原理​

ReAct 模式的核心思想是：Agent 在每一步都会先"思考"（Reason），然后选择一个"行动"（Act），观察结果后再进行下一轮思考。这个循环持续进行，直到 Agent 能够给出最终答案。

一个典型的 ReAct 执行流程：

用户问题：找出目前最受欢迎的无线耳机并确认是否有货

思考：流行度是时间敏感的，我需要使用搜索工具获取最新信息。
行动：调用 search_products("wireless headphones")
观察：搜索结果显示 Sony WH-1000XM5 最受欢迎...

思考：我需要确认排名第一的产品是否有库存才能回答。
行动：调用 check_inventory("WH-1000XM5")
观察：库存充足，有 50 件...

思考：我已经获得了最受欢迎的型号和库存状态，可以回答用户的问题了。
行动：输出最终答案

ReAct 的优势​

  • 可解释性强：每一步推理过程都清晰可见，便于调试和审计
  • 自我纠错：能够根据观察结果调整策略，提高任务完成率
  • 灵活适应：可以处理各种意外情况和边缘场景
  • 工具链式调用：支持复杂的多步骤任务执行

LangChain 1.0 中的 ReAct 实现​

在 LangChain 1.0 中，create_agent 默认采用 ReAct 模式，Agent 会在模型节点和工具节点之间循环：

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def search_products(query: str) -> str:
    """搜索产品信息"""
    return f"找到产品：{query}，最受欢迎的是 Sony WH-1000XM5"

@tool
def check_inventory(product_id: str) -> str:
    """检查库存"""
    return f"{product_id} 库存充足"

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[search_products, check_inventory],
    system_prompt="你是一个购物助手，帮助用户查找产品信息。"
)

# 流式输出可以看到完整的 ReAct 过程
for chunk in agent.stream({
    "messages": [{"role": "user", "content": "帮我找最受欢迎的无线耳机"}]
}):
    # 注意：节点名是 "model" 而不是 "agent"
    if "model" in chunk:
        for msg in chunk["model"]["messages"]:
            if hasattr(msg, "tool_calls") and msg.tool_calls:
                print(f"[思考] 调用工具: {msg.tool_calls[0]['name']}")
            elif msg.content:
                print(f"[回答] {msg.content}")
    elif "tools" in chunk:
        for msg in chunk["tools"]["messages"]:
            print(f"[观察] {msg.content}")

使用 LangGraph 构建自定义 ReAct Agent​

对于更复杂的场景，可以直接使用 LangGraph 的 StateGraph 构建：

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode, tools_condition
from typing import Annotated
from langgraph.graph.message import add_messages
from typing import TypedDict

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

@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果：{query} 的相关信息..."

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{e}"

tools = [search, calculator]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 构建自定义 ReAct Agent
graph_builder = StateGraph(State)
graph_builder.add_node("tools", ToolNode(tools))
graph_builder.add_node("chatbot", lambda state: {
    "messages": [llm.bind_tools(tools).invoke(state['messages'])]
})
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()

result = graph.invoke({
    "messages": [{"role": "user", "content": "北京和上海的人口总和是多少？"}]
})
print(result["messages"][-1].content)

Function Calling 模式​

Function Calling 是 OpenAI 推出的一种让 LLM 调用外部函数的机制。在 LangGraph 中，工具调用被统一整合到 Agent 架构中，底层使用的就是 Function Calling 机制。

工作原理​

当 LLM 判断需要调用工具时，它会返回一个结构化的工具调用请求，包含工具名称和参数。应用执行工具后，将结果返回给 LLM 进行下一步处理。

工具定义格式​

工具使用 JSON Schema 定义参数结构，LangChain 会自动从 @tool 装饰器的函数签名中提取：

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    """天气查询参数"""
    city: str = Field(description="城市名称，如：北京、上海")
    unit: str = Field(default="celsius", description="温度单位")

@tool(args_schema=WeatherInput)
def get_weather(city: str, unit: str = "celsius") -> str:
    """
    获取指定城市的天气信息
    
    参数：
    - city: 城市名称
    - unit: 温度单位（celsius 或 fahrenheit）
    """
    weather_data = {
        "北京": {"temp": 22, "condition": "晴"},
        "上海": {"temp": 26, "condition": "多云"},
    }
    data = weather_data.get(city, {"temp": 20, "condition": "未知"})
    
    if unit == "fahrenheit":
        temp = data["temp"] * 9/5 + 32
    else:
        temp = data["temp"]
    
    return f"{city}：{data['condition']}，温度 {temp}°{unit[0].upper()}"

结构化输出​

使用 response_format 参数可以让 Agent 返回结构化的结果：

from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel, Field
from typing import List

class AnalysisResult(BaseModel):
    """分析结果结构"""
    summary: str = Field(description="内容摘要")
    key_points: List[str] = Field(description="关键点列表")
    sentiment: str = Field(description="情感倾向：positive/neutral/negative")

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    response_format=AnalysisResult
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "分析这段文本的情感：今天天气真好！"}]
})

# 结果会包含结构化的 AnalysisResult
print(result["structured_response"])

中断与人工确认​

在某些场景下，我们需要在执行关键操作前让用户确认。LangChain 1.0 推荐使用 middleware 的 after_model 方法实现这个功能：

from langchain.agents import create_agent, Middleware
from langgraph.checkpoint.memory import MemorySaver
from langchain.tools import tool

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送电子邮件"""
    return f"邮件已发送至 {to}"

@tool
def delete_file(file_path: str) -> str:
    """删除文件"""
    return f"已删除文件：{file_path}"

checkpointer = MemorySaver()

class HumanApprovalMiddleware(AgentMiddleware):
    """人工确认中间件"""
    
    def __init__(self, require_approval_tools: list):
        self.require_approval_tools = require_approval_tools
    
    def after_model(self, state, context):
        """在模型调用后检查是否需要确认"""
        messages = state.get("messages", [])
        if not messages:
            return
        
        last_msg = messages[-1]
        if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
            for tc in last_msg.tool_calls:
                if tc["name"] in self.require_approval_tools:
                    # 设置中断标志
                    return {"interrupt": True, "pending_tool": tc}

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[send_email, delete_file],
    checkpointer=checkpointer,
    middleware=[HumanApprovalMiddleware(["send_email", "delete_file"])],
    interrupt_before=["tools"]  # 在工具执行前中断
)

config = {"configurable": {"thread_id": "session-001"}}

# 第一次调用，Agent 会规划要执行的工具，但不会真正执行
result = agent.invoke(
    {"messages": [{"role": "user", "content": "发送邮件给 [email protected]，主题是测试"}]},
    config=config
)

# 检查下一步要执行的操作
state = agent.get_state(config)
print(f"待执行的工具调用: {state.next}")

# 如果用户确认，继续执行
if user_confirms():
    result = agent.invoke(None, config=config)  # 传入 None 继续之前的执行

错误处理策略​

Agent 在执行过程中可能遇到各种错误，需要设计健壮的错误处理机制。

使用 ToolNode 的内置错误处理​

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

@tool
def risky_operation(param: str) -> str:
    """可能失败的操作"""
    if param == "bad":
        raise ValueError("无效参数")
    return f"成功处理：{param}"

# 默认错误处理：捕获错误并返回错误消息
tool_node = ToolNode([risky_operation])

# 自定义错误消息
tool_node = ToolNode(
    [risky_operation],
    handle_tool_errors="操作执行失败，请检查输入参数。"
)

重试机制​

from tenacity import retry, stop_after_attempt, wait_exponential

@tool
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def call_external_api(endpoint: str) -> str:
    """调用外部 API（带重试）"""
    import requests
    response = requests.get(endpoint, timeout=10)
    response.raise_for_status()
    return response.text

优雅降级​

@tool
def search_with_fallback(query: str) -> str:
    """搜索信息（带降级策略）"""
    try:
        # 尝试主搜索源
        return primary_search(query)
    except Exception as e:
        # 降级到备用搜索源
        try:
            return fallback_search(query)
        except Exception:
            return f"搜索服务暂时不可用，请稍后重试"

性能优化​

并行工具调用​

当模型返回多个工具调用时，ToolNode 会自动并行执行：

from langgraph.prebuilt import create_react_agent, ToolNode
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取天气"""
    return f"{city}：晴"

@tool
def get_news(topic: str) -> str:
    """获取新闻"""
    return f"{topic} 相关新闻..."

# ToolNode 默认并行执行多个工具调用
agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[get_weather, get_news]
)

# 如果模型同时调用两个工具，它们会并行执行
result = agent.invoke({
    "messages": [{"role": "user", "content": "北京天气怎么样？有什么科技新闻？"}]
})

缓存机制​

对于重复的查询，可以添加缓存：

from functools import lru_cache

@tool
@lru_cache(maxsize=100)
def cached_search(query: str) -> str:
    """带缓存的搜索"""
    # 相同的查询会直接返回缓存结果
    return expensive_search(query)

流式输出​

使用流式输出提升用户体验：

async def stream_agent_response(agent, user_input: str):
    """流式输出 Agent 响应"""
    async for chunk in agent.astream({
        "messages": [{"role": "user", "content": user_input}]
    }):
        if "agent" in chunk:
            for msg in chunk["agent"]["messages"]:
                if msg.content:
                    yield msg.content

小结​

设计一个优秀的 Agent 架构需要考虑多个方面：

  • LangGraph 采用 Graph-based 架构，提供更好的可控性和可扩展性
  • create_agent 和 create_react_agent 都是有效的 Agent 创建方式，前者支持 middleware，后者更简洁
  • 理解 ReAct 模式的工作原理，这是 Agent 的核心执行模式
  • Function Calling 是工具调用的底层机制
  • 使用 checkpointer 和 store 管理短期和长期记忆
  • 通过 interrupt_before 实现人工确认
  • 使用 Middleware（仅 create_agent）的 wrap_tool_call 方法处理工具执行错误
  • 通过自定义 State 扩展 Agent 能力

下一章我们将深入学习工具系统的设计与实现。

参考资料​