XBSTACK XBSTACK
小白 / Xiaobai

小白 / Xiaobai

开发者 · 产品构建者

持续构建 AI 工程系统、开发者工具与长期数字资产。

关于作者与 XBSTACK →
LangChain v1 使用 create_agent、Middleware、Memory 与 Human-in-the-loop 构建生产 Agent

LangChain v1 实战:用 create_agent、Middleware、Memory 与 HITL 构建 Agent

LangChain v1 Agent 怎么做?本文按当前 create_agent 主线拆解工具调用、Middleware、短期记忆、Runtime Context、Human-in-the-loop 与 LangGraph 持久化边界,替代旧 AgentExecutor 教程。

发布 · 2026-04-255 分钟阅读XBSTACK 原创
#AI Agent#LangChain#LangChain v1#LangGraph#Python#Middleware#Human-in-the-loop

直接答案:如果你在 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=Truemax_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 重投和网络超时仍可能造成外部副作用重复。支付、发布、发信、写数据库等操作必须使用业务幂等键和唯一约束。

继续阅读

专题入口 / LangGraph Hub

继续按生产级 LangGraph 路线读,不再重复看泛入门

这一类文章统一沉淀到 LangGraph 专题页,按状态隔离、Checkpointer、HITL、失败恢复、Observability、Supervisor/Worker、Subgraph 和 Memory 顺序阅读。

继续阅读

返回专题 →
OpenAI Agents SDK Tool Approval 如何恢复?RunState 跨进程与 v0.19.3 流式 Resume 实测OpenAI Agents SDK RunState 如何恢复 Tool Approval?本文对比 openai-agents 0.18.3 与 0.19.3,实测跨进程批准/拒绝、流式 Resume 丢失已批准 Tool Output 的回归与修复,并验证重复投递、业务幂等和 Context 秘密边界。AutoGen 实战教程:AgentChat、Teams、Termination 与 v0.2 迁移边界AutoGen 当前怎么用?本文按 AgentChat/Core/Extensions 分层讲解 AssistantAgent、RoundRobinGroupChat、SelectorGroupChat、Termination Conditions、UserProxyAgent、状态恢复与 v0.2 迁移边界。OpenAI Agents SDK 重复 Tool 名称:为什么后注册工具会覆盖前一个?OpenAI Agents SDK 重复 Tool 名称:实测 openai-agents 0.19.2:两个 FunctionTool 使用同名 lookup 时,SDK 校验不会报错,Agent 仍把两个工具交给模型,而本地分发表只保留后注册工具。本文给出离线复现、风险边界、启动前校验和修复方案。AI Agent 协议与框架选型:MCP、Function Calling、A2A、LangGraph、AutoGen、CrewAI 怎么选?AI Agent 协议与框架选型:系统梳理 AI Agent 开发中的协议与框架选型,覆盖 Function Calling、MCP、A2A、LangGraph、AutoGen、CrewAI、LangChain、自研 Workflow、多智能体协作、工具调用、状态管理和生产化边界,帮助开发者根据场景选择合适技术栈。

AI 工程周报

只发真正改变工程判断的变化、故障、实验和新资产。

评论与补充证据

参与讨论

问题、验证与勘误

登录后可发表评论。所有新评论先进入审核;审核期间仅评论者本人和管理员可见,通过后才公开。

登录评论 审核后公开
正在加载评论区…