跳到主要内容

create_agent

概述​

create_agent 是 LangChain 1.0 的上层统一入口，内部依赖 LangGraph。调用时自动构建基于 ReAct 范式的图结构，包含 Agent 决策、工具调用、状态更新等核心节点。

ReAct 执行循环​

ReAct（Reasoning + Acting）是 Agent 的核心执行机制，强调"推理—行动—观察"的闭环。

执行流程​

用户输入 → Thought → Action → Observation → 循环决策 → 最终答案
步骤说明
Thought大模型思考，决定下一步行动
Action选择工具并构造输入参数
Observation工具执行返回的结果
循环将观察结果纳入上下文，继续推理

终止条件​

  • 达到最终答案
  • 迭代上限（recursion_limit）
  • 时间上限
  • 发生错误

代码示例​

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

# 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    data = {"北京": "晴，25°C", "上海": "多云，28°C"}
    return f"{city}：{data.get(city, '未知')}"

@tool
def add(a: float, b: float) -> float:
    """计算两个数的和"""
    return a + b

# 创建 Agent
agent = create_agent(
    model=model,
    tools=[get_weather, add],
    system_prompt="你是一个多功能助手"
)

# 执行任务（流式输出观察推理过程）
config = {"configurable": {"thread_id": "demo"}}
for step in agent.stream(
    {"messages": [{"role": "user", "content": "查询北京和上海天气，计算温度差"}]},
    config=config,
    stream_mode="values"
):
    message = step["messages"][-1]
    message.pretty_print()

函数签名​

create_agent 通过三要素 + 三扩展的极简抽象重构了 Agent 开发范式：

分类要素作用
三要素模型（Model）、工具（Tools）、提示词（System Prompt）构成 Agent 的"灵魂"，决定它能思考什么、能做什么
三扩展中间件（Middleware）、内存管理（Memory）、状态管理（State）构建 Agent 的"神经系统"，提供生产级可靠性
agent = create_agent(
    model,                    # 模型（必填）
    tools,                    # 工具列表（必填）
    system_prompt=None,       # 系统提示词
    middleware=[],            # 中间件列表
    checkpointer=None,        # 短期记忆
    store=None,              # 长期记忆
    state_schema=None,        # 扩展状态
    context_schema=None,      # 动态上下文
    response_format=None     # 结构化输出
)

参数详解​

参数类型必填作用
modelstr/实例✅推理引擎
toolslist✅执行能力
system_promptstr-行为准则
middlewarelist-功能扩展
checkpointerSaver-短期记忆
storeStore-长期记忆
state_schemaTypedDict-扩展状态
context_schemaTypedDict-动态上下文
response_formatBaseModel-结构化输出

基础示例​

from langchain.agents import create_agent

agent = create_agent(
    model=model,                    # 模型
    tools=[order_query_tool],       # 工具
    system_prompt="你是一个订单查询助手，能够查询订单状态和明细。" , # 系统提示
    middlewares=[order_query_middleware],                    # 中间件
    checkpointer=checkpointer,      # 检查点短期记忆
    store=store,                    # 状态存储长期记忆
    state_schema=OrderQueryState,   # 扩展状态（如需要）
    context_schema=AgentContext,    # 上下文状态（如需要）
    response_format=ResponseModel   # 结构化输出（如需要）
)

# ============ 限制最大 3 次循环 ============
config = {
    "configurable": {"thread_id": "limit_demo"}, # 限制thread_id 线程ID
    "recursion_limit": 3  # 最多 3 次迭代，或使用中间件进行精确跟踪和终止循环
}

result = agent.invoke(
        {"messages": [{"role": "user", "content": "LangChain 1.0 发布日期"}]},
        config=config
    )

限制循环次数​

config = {
    "configurable": {"thread_id": "demo"},
    "recursion_limit": 3  # 最多 3 次迭代
}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "问题"}]},
    config=config
)

配置限速器​

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=5,
    check_every_n_seconds=1.0
)

model = init_chat_model(
    model="deepseek-chat",
    provider="deepseek",
    rate_limiter=rate_limiter
)