小白 / Xiaobai
开发者 · 产品构建者
持续构建 AI 工程系统、开发者工具与长期数字资产。
关于作者与 XBSTACK →
LangChain v1 实战:用 create_agent、Middleware、Memory 与 HITL 构建 Agent
LangChain v1 Agent 怎么做?本文按当前 create_agent 主线拆解工具调用、Middleware、短期记忆、Runtime Context、Human-in-the-loop 与 LangGraph 持久化边界,替代旧 AgentExecutor 教程。
直接答案:如果你在 2026 年新写 LangChain Agent,应该从 create_agent 开始,而不是再照着旧教程搭 AgentExecutor。LangChain v1 已把 create_agent 作为标准 Agent API,它运行在 LangGraph runtime 上,并通过 Middleware、Checkpointer、Runtime Context 和 Human-in-the-loop 把工具调用、记忆、权限与恢复组织到同一个执行模型里。
旧的 AgentExecutor 思路并不是“历史上完全错误”,但它已经不适合作为当前教程的主线。继续把 handle_parsing_errors=True、max_iterations 和 scratchpad 当成 2026 的核心教学,会让新项目从一开始就背上旧 API 心智模型。
最小可用 Agent:create_agent + 强类型工具
当前最小结构可以保持很简单:
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.tools import tool
class WeatherInput(BaseModel):
location: str = Field(description="City name")
@tool(args_schema=WeatherInput)
def get_weather(location: str) -> str:
"""Return weather data from the application's trusted weather service."""
return f"weather:{location}"
agent = create_agent(
model="your-model",
tools=[get_weather],
system_prompt=(
"Use tools only when needed. Never invent tool results. "
"Ask for clarification when required inputs are missing."
),
)
result = agent.invoke({
"messages": [{"role": "user", "content": "贵阳今天适合徒步吗?"}]
})
这里真正需要治理的不是“循环写几行”,而是工具 Schema、权限、错误分类和外部副作用。
create_agent 和 LangGraph 到底是什么关系
LangChain v1 的 create_agent 并不是另起一套运行时。官方当前文档明确:LangChain agents 构建在 LangGraph 上。
可以把两层理解为:
- LangChain:高层 Agent API、模型/工具集成、Middleware、结构化输出;
- LangGraph:底层状态、执行、持久化、interrupt、streaming 与自定义图拓扑。
所以简单 Agent 不需要一上来手写 StateGraph;当业务需要分类后分流、并行 fan-out、自定义恢复语义或多个 deterministic 节点时,再把 agent 作为一个节点/子图嵌进更大的 LangGraph。
Middleware 才是当前生产控制面的重点
LangChain v1 把很多过去散落在 AgentExecutor 参数、Callback 和自定义 Wrapper 里的控制能力收敛到 Middleware。
当前常见能力包括:
- 动态 system prompt;
- 会话总结与上下文裁剪;
- Model/Tool call limit;
- Tool retry / Model retry;
- Model fallback;
- PII 检测与脱敏;
- Human-in-the-loop;
- 自定义 before_model / after_model / wrap_tool_call 等控制点。
这意味着生产 Agent 的问题不应该只问“怎么防死循环”,而应该问:模型最多调用几次、哪些工具允许重试、哪些错误不能重试、哪些工具必须人工批准、哪些数据禁止进入模型上下文。
记忆:短期 thread state 和长期 memory 不要混
LangChain 当前把线程级短期记忆放进 Agent state,并通过 LangGraph checkpointer 持久化。
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_agent(
model="your-model",
tools=[get_weather],
checkpointer=checkpointer,
)
config = {"configurable": {"thread_id": "user-42:trip-1"}}
agent.invoke(
{"messages": [{"role": "user", "content": "我周末想去徒步"}]},
config=config,
)
InMemorySaver 适合本地示例。生产环境应该使用数据库型 checkpointer,例如 Postgres,并建立 thread_id 生命周期、租户隔离和清理策略。
跨 thread 的长期用户偏好、账户信息或长期业务记忆则应该使用 Store 或应用自己的数据库。不要把“所有历史消息一直塞回 Prompt”当成 memory 设计。
Runtime Context:工具依赖不要硬编码进全局变量
当前 LangChain Runtime 可以向工具和 Middleware 注入 context、store 与执行信息。这个机制适合传递 user_id、数据库连接、租户配置和运行时依赖。
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
@dataclass
class Context:
user_id: str
@tool
def load_profile(runtime: ToolRuntime[Context]) -> str:
"""Load the current user's approved profile fields."""
return f"user:{runtime.context.user_id}"
agent = create_agent(
model="your-model",
tools=[load_profile],
context_schema=Context,
)
这样 Tool 逻辑更容易测试,也不会把用户身份、数据库对象或密钥硬塞进 Prompt。
Human-in-the-loop:高风险 Tool 在执行前暂停
对于发送邮件、写文件、执行 SQL、发布、删除、支付等动作,应在模型提出 Tool Call 后、真正执行前做审批。
LangChain 当前提供 HumanInTheLoopMiddleware。它可以按工具名称决定是否 interrupt,并让用户做 approve / edit / reject。
但要注意:框架的 interrupt 只是运行暂停能力。生产审批还应该保存:
- approval_id;
- 谁批准;
- 批准了什么参数;
- Tool Schema/业务版本;
- 过期时间;
- 幂等键;
- 最终执行结果。
否则“能够 resume”仍然不等于“审批系统安全”。
错误处理:不要再把 handle_parsing_errors 当核心答案
现代 Tool Calling 已经大量使用结构化 Schema。错误处理应该按层分类:
| 错误层 | 建议 |
|---|---|
| 参数校验失败 | 把可修正字段反馈给模型,限制重试次数 |
| 429 / 临时网络失败 | 有界退避,遵守服务端 Retry-After |
| 权限拒绝 | 不重试,返回明确授权错误 |
| 外部写入超时 | 先查幂等结果,再决定是否重试 |
| 模型循环无进展 | Model/Tool call limit + 任务进度判断 |
| 高风险动作 | HITL,不能靠模型自评置信度自动放行 |
什么时候只用 create_agent,什么时候直接用 LangGraph
优先 create_agent:
- 单 Agent + 多工具;
- 工具循环;
- 常规 RAG;
- Middleware 可以覆盖的权限、重试、总结与 HITL。
直接进入 LangGraph:
- 明确的多阶段状态机;
- 并行分支与 fan-in;
- 多 Agent 编排;
- 自定义 checkpoint / resume 边界;
- 长时间业务流程和复杂补偿路径。
关键不是框架层级越低越专业,而是只在默认 Agent loop 不够表达业务拓扑时,才增加图复杂度。
FAQ
AgentExecutor 代码需要马上全部重写吗?
不需要为了形式立即重写稳定生产系统,但新功能应先评估 LangChain v1 的 create_agent、Middleware 和当前持久化模型。迁移优先处理那些依赖旧 parser、旧 memory、旧 callback 或无法表达 HITL/恢复边界的部分。
LangChain Agent 能自动保证工具只执行一次吗?
不能。重试、Worker 重投和网络超时仍可能造成外部副作用重复。支付、发布、发信、写数据库等操作必须使用业务幂等键和唯一约束。
继续阅读
- AI Agent 全栈指南
- LangGraph 失败恢复:Retry、Timeout 与 Fallback
- LangGraph Human-in-the-loop 审批流
- AI Agent Tool Authorization Policy Gate
继续按生产级 LangGraph 路线读,不再重复看泛入门
这一类文章统一沉淀到 LangGraph 专题页,按状态隔离、Checkpointer、HITL、失败恢复、Observability、Supervisor/Worker、Subgraph 和 Memory 顺序阅读。
继续阅读
返回专题 →AI 工程周报
只发真正改变工程判断的变化、故障、实验和新资产。
参与讨论
问题、验证与勘误
登录后可发表评论。所有新评论先进入审核;审核期间仅评论者本人和管理员可见,通过后才公开。