LangChain + LangGraph 核心组件：从 Agent 封装到可控工作流

很多人第一次接触 LangChain看到的公式是用户消息 → Model → Tools → Model → 最终答案这条链路没有错但它只解释了 Agent 的“主循环”没有解释谁负责切换模型、校验权限、保存对话、处理工具异常以及在高风险操作前暂停等待人工审批。更完整的理解应该是Agent Model HarnessModel 负责理解、推理和决定下一步动作Harness Tools Prompt Middleware Context State Memory RuntimeLangChain 负责把这些常用能力封装成容易使用的 Agent APILangGraph 则把 Agent 进一步展开成状态、节点、边、分支、循环和中断让开发者精确控制执行过程。本文会先拆解 LangChain 的 Model、Tool、response format、middleware 和 runtime再用 State、Node、Edge 解释 LangGraph最后用 Python 和真实 OpenAI API 跑通一个“识别城市 → 查询天气”的最小 ReAct Agent。本文 API 根据 2026 年 8 月 LangChain 与 LangGraph 官方 Python 文档整理。两个框架更新较快旧教程中的initialize_agent、AgentExecutor等写法不代表当前推荐入口。一、先看入口create_agent 到底创建了什么当前 LangChain 构建 Agent 的核心入口是create_agentfrom langchain.agents import create_agentagent create_agent( modelmodel, toolstools, system_prompt你是一个天气助手, middlewaremiddleware, response_formatWeatherAnswer, checkpointercheckpointer, context_schemaUserContext,)result agent.invoke( {messages: [{role: user, content: 杭州今天天气如何}]}, config{configurable: {thread_id: weather-001}}, context{user_id: u-1001, role: member},)这里的create_agent并不是立即执行一次模型调用而是组装出一套可执行框架。真正开始运行的是agent.invoke()。配置阶段 model tools prompt middleware response_format checkpointer context_schema ↓create_agent 编译 Agent 执行图 ↓agent.invoke 注入本轮 messages、config 和 context ↓运行阶段 Model 判断 → Tool 执行 → Observation 回填 → Model 再判断 ↓结束阶段 返回 messages、结构化结果和更新后的状态一个容易忽略的事实是create_agent底层运行在 LangGraph 上。它返回的不是一个简单函数而是一张已经编译好的图因此天然支持invoke、stream、状态持久化和中断恢复。关键洞察LangChain 的create_agent是预制好的 Agent HarnessLangGraph 是支撑这套 Harness 的图运行时也是需要精细控制流程时可以直接使用的底层编排能力。二、Model用统一接口屏蔽不同模型供应商Model 是 Agent 的推理引擎。它负责阅读消息和工具定义判断应该直接回答还是生成一个或多个 tool call。LangChain 的价值不是“提供一个模型”而是为不同 Provider 提供相对统一的调用接口from langchain.chat_models import init_chat_modelmodel init_chat_model( openai:gpt-4.1-mini, temperature0, timeout30, max_retries3,)如果换成 AnthropicAgent 上层调用方式基本不变model init_chat_model( anthropic:claude-sonnet-4-6, temperature0, timeout30, max_retries3,)Model 层可以拆成三部分部分作用示例Provider提供模型服务的厂商或平台OpenAI、Anthropic、Google、AzureModel Object统一的模型调用对象init_chat_model()、ChatOpenAI()调用策略控制稳定性与流量分配temperature、timeout、retry、route、fallbacktemperature、timeout和max_retries属于模型对象的基础配置动态路由、fallback、缓存等更适合放在 middleware 的wrap_model_call中。因为路由通常需要结合本轮状态判断而不是在 Agent 创建时永久写死。例如普通问答走小模型长上下文或高风险任务走强模型主模型超时后重试仍失败再切换备用模型。关键洞察Model 统一接口解决“怎么调用模型”middleware 解决“这一轮到底调用哪个模型以及失败后怎么办”。三、Tool让普通函数成为模型可选择的动作Tool 的本质是把一个普通函数包装成模型能够理解和调用的动作。一个完整 Tool 通常包含四部分字段模型如何使用天气工具示例name区分工具生成 tool call 时指定名称get_weatherdescription判断什么时候调用、能解决什么问题查询指定城市的天气schema生成参数确定字段、类型和必填项city: strfunction真正执行查询、写入或计算调用天气 API 或数据库最简单的写法是使用toolfrom langchain.tools import tooltooldef get_weather(city: str) - dict: 查询指定城市的当前天气。仅在已经获得标准城市名后调用。 return { city: city, temperature_c: 26, condition: 晴, }函数名默认成为 Tool namedocstring 成为 description类型注解生成参数 schema。复杂参数可以用 Pydantic 显式定义from pydantic import BaseModel, Fieldclass WeatherInput(BaseModel): city: str Field(description标准城市名例如杭州) unit: str Field(defaultcelsius, description温度单位)这里需要纠正一个常见误解模型通常看不到函数源码也不会阅读 Python 或 TypeScript 里的业务实现。模型能够看到 - name - description - 参数 schema模型通常看不到 - function source code - Python 内部实现 - TypeScript 函数内部逻辑 - Zod 校验器的源码在 TypeScript 中常用 Zod 定义 schema但模型接收到的是转换后的 JSON Schema不是 Zod 源码。真正决定工具是否容易被正确调用的是清晰的名称、准确的 description 和不过度复杂的参数结构。还有一个工程细节某些参数只供运行时使用例如ToolRuntime。这些参数由系统注入不会暴露给模型因此可以安全地传递用户身份、当前 State、Store 或 tool call ID。四、response_format结构化输出不是“提醒模型返回 JSON”如果 Agent 的结果还要交给程序消费仅让模型“请返回 JSON”通常不够稳定。模型可能加 Markdown 围栏、漏字段、写错类型甚至在 JSON 前后补解释。LangChain 的response_format用 schema 定义最终结果from pydantic import BaseModel, Fieldclass WeatherAnswer(BaseModel): city: str Field(description标准城市名) temperature_c: int Field(description摄氏温度) condition: str Field(description天气状况) suggestion: str Field(description一句出行建议)创建 Agent 时直接传入agent create_agent( modelmodel, tools[resolve_city, get_weather], response_formatWeatherAnswer,)调用结束后从structured_response读取result agent.invoke({ messages: [{role: user, content: 西湖那边今天热吗}]})print(result[structured_response])LangChain 会根据模型能力选择 ProviderStrategy 或 ToolStrategy模型原生支持结构化输出时优先使用 ProviderStrategy否则可以通过工具调用策略生成并校验结构。“返回 JSON” 只是在 Prompt 中提出文本要求 程序仍要自行解析和校验response_format 用 Schema 定义字段、类型和约束 框架负责结构化生成与校验 结果作为 typed data 交给下游程序关键洞察Tool schema 约束的是“Agent 调用动作时输入什么”response format 约束的是“Agent 最终向业务返回什么”。两者不要混为一谈。五、MiddlewareAgent 的执行控制面如果 Model 是大脑Tools 是手脚middleware 就是控制面。一次 Agent 调用可能经历多轮模型和工具调用User Message → Model Call → Tool Call → Tool Result → Model Call → Tool Call → Tool Result → Model Call → Final AnswerPrompt 很难可靠地承担身份校验、超时、重试、工具权限、成本限制和日志追踪。middleware 可以在每一步做确定性的检查、改写、拦截和兜底。5.1 节点式 hooks在阶段前后执行一次节点式 hooks 适合执行阶段级逻辑Hook触发位置常见用途before_agent整次 Agent 启动前身份校验、初始化运行状态before_model每次 Model 调用前裁剪消息、改写 Prompt、统计调用次数after_model每次 Model 返回后检查输出、识别异常工具调用、统计 Tokenafter_agentAgent 结束后保存结果、写 Trace、记录业务指标注意before_model和after_model可能执行多次因为 ReAct Agent 会反复回到模型节点。5.2 包裹式 hooks接管某一次调用wrap hooks 类似在目标调用外面包一层控制器Hook包裹对象适合处理wrap_model_call一次模型调用模型路由、retry、fallback、cache、动态工具集合wrap_tool_call一次工具调用工具权限、参数过滤、日志、超时、失败兜底例如工具异常时返回一条可理解的 ToolMessage而不是让整个 Agent 直接崩溃from langchain.agents.middleware import wrap_tool_callfrom langchain.messages import ToolMessagewrap_tool_calldef handle_tool_error(request, handler): try: return handler(request) except Exception as exc: return ToolMessage( contentf工具暂时不可用{exc}, tool_call_idrequest.tool_call[id], )生产环境也可以直接使用ModelRetryMiddleware、ToolRetryMiddleware等预制中间件不必为通用逻辑重复造轮子。节点式 hook 在某个阶段前后增加处理节点wrap hook 接管一次具体调用 可以决定是否执行、执行几次、换谁执行以及异常如何返回middleware 并不是另一个独立运行时。它最终也会成为create_agent所编译 LangGraph 中的一部分。六、Runtime一次 invoke 运行需要哪些数据agent.invoke()不只是传一段 Prompt。一次完整运行通常有三类输入输入生命周期用途messages / state会随流程更新用户消息、模型回答、工具结果、业务状态config控制本次执行thread_id、callbacks、tags、metadatacontext本次运行只读业务信息user_id、role、tenant_id、feature_flags可以把它们记成一句话State流程正在处理和修改的数据包Config框架如何运行这一次任务Context业务系统告诉 Agent“你正在为谁、以什么权限运行”6.1 context 不是对话记忆context 适合存放本次运行依赖但不应该由模型随意修改的信息from dataclasses import dataclassdataclassclass UserContext: user_id: str role: str tenant_id: str创建 Agent 时声明context_schema调用时传入 contextagent create_agent( modelmodel, toolstools, context_schemaUserContext,)agent.invoke( {messages: [{role: user, content: 查询杭州天气}]}, contextUserContext( user_idu-1001, rolemember, tenant_idt-01, ),)Tool 或 middleware 可以通过 Runtime 读取这些信息执行租户隔离或权限判断。6.2 checkpointer thread_id 才能保存多轮状态如果希望第二次调用记住第一次对话需要给 Agent 配置 checkpointer并在调用时使用稳定的thread_idfrom langgraph.checkpoint.memory import InMemorySavercheckpointer InMemorySaver()agent create_agent( modelmodel, toolstools, checkpointercheckpointer,)config {configurable: {thread_id: weather-001}}相同thread_id指向同一条会话状态换一个thread_id就是另一条独立会话。组件保存什么适合场景Checkpointer单个 thread 的 State 快照多轮对话、故障恢复、人工审批Store跨 thread 的长期数据用户偏好、共享知识、长期记忆示例中的InMemorySaver只适合本地演示。进程退出后内容会消失生产环境应换成数据库支持的持久化 Checkpointer。七、LangChain 与 LangGraph 到底有什么区别LangChain 和 LangGraph 不是二选一也不是简单的“低级框架与高级框架”。两者解决的是不同层面的问题。对比维度LangChainLangGraph核心关注Model、Tools、middleware、structured outputState、Node、Edge、分支、循环、暂停、调度开发方式使用封装好的高层 API 快速搭建 Agent显式设计状态图和执行路径默认决策模型根据上下文决定是否调用工具代码和状态共同决定走哪条边灵活度常见 Agent 模式开箱即用可精确控制每个节点和状态变化适合场景问答、检索、工具型助手、标准 ReAct多分支流程、审批、长任务、复杂恢复逻辑主要成本深度定制时可能受到预制循环限制需要自行设计 State、Node、Edge 和测试优先选择 LangChain 目标是快速搭建标准工具调用 Agent 主流程就是 Model ↔ Tools 循环 使用 middleware 已能满足控制要求优先直接使用 LangGraph 业务包含多个确定性分支 需要显式循环、并行、暂停或人工审批 需要精确保存和恢复每一步状态 不希望所有流程选择都交给模型实际项目经常组合使用用create_agent构建一个标准 Agent再把它作为 LangGraph 中的一个 Node 或 Subgraph。关键洞察LangChain 关注“一个 Agent 需要哪些能力”LangGraph 关注“这些能力按照什么状态和路径运行”。八、State Node Edge把 Agent 展开成一张图LangGraph 最重要的三个概念是 State、Node 和 EdgeState运行时携带的数据包Node读取 State、执行处理、返回 State 更新的函数Edge决定执行完当前 Node 后去哪里State Node Edge Agent Flow8.1 State流程携带的数据包天气 Agent 的 State 可以包含from typing_extensions import TypedDictfrom langgraph.graph import MessagesStateclass WeatherState(MessagesState): city: str | None weather: dict | None retry_count: int其中State 字段保存内容messages用户 Message、模型回答、tool call、ToolMessagecity标准化后的城市名weather工具返回的天气结构retry_count查询失败后的重试次数State 不是一次性 Prompt而是整张图运行时持续携带和更新的数据。8.2 Node流程中的处理步骤Node 本质上是函数。它读取当前 State执行一个明确步骤并返回局部更新def resolve_city_node(state: WeatherState): city parse_city(state[messages][-1].content) return {city: city}def query_weather_node(state: WeatherState): weather weather_api(state[city]) return {weather: weather}常见 Node 包括调用模型查询工具或数据库校验结果人工审核生成最终答案Node 应该有清晰职责。一个节点同时查询数据库、调用模型、写文件并发送通知会让状态恢复和失败重试变得困难。8.3 Edge决定下一步去哪里普通边是固定路线builder.add_edge(resolve_city, query_weather)表示resolve_city执行完后一定进入query_weather。条件边根据 State 动态选择路径def route_after_model(state): last_message state[messages][-1] if last_message.tool_calls: return tools return end这正是标准 ReAct Agent 的核心条件如果模型返回 tool_call 进入 Tools Node 执行工具并把结果写回 messages 再回到 Model Node如果模型没有返回 tool_call 说明模型准备给出最终答案 流程进入 ENDModel 和 Tools 之间的循环并不是神秘的“智能涌现”而是条件边和回边共同形成的执行路径。九、Interrupt让 Agent 在关键动作前停下来当流程准备合并代码到 main 分支时完全自动执行往往风险过高。更合理的流程是先查询 CR 状态、生成审核建议再暂停等待人工决定。在这个案例中State repo、branch、cr_status、review_summary、approvedNode query_cr、generate_review、human_approval、merge_main、reject普通 Edge query_cr → generate_review → human_approval条件 Edge approved true → merge_main approved false → rejectLangGraph 使用interrupt()暂停from langgraph.types import interruptdef human_approval(state: ReviewState): approved interrupt({ question: 是否同意合并到 main, cr_status: state[cr_status], review_summary: state[review_summary], }) return {approved: approved}为了恢复流程必须同时具备编译图时配置 Checkpointer。首次执行时提供稳定的thread_id。恢复时使用同一个thread_id。通过Command(resume...)传入人工结果。from langgraph.checkpoint.memory import InMemorySaverfrom langgraph.types import Commandgraph builder.compile(checkpointerInMemorySaver())config {configurable: {thread_id: merge-20260802-001}}# 第一次运行在 interrupt 处暂停paused graph.invoke(initial_state, configconfig)# 人工确认后恢复True 会成为 interrupt() 的返回值result graph.invoke(Command(resumeTrue), configconfig)需要特别注意恢复时包含interrupt()的 Node 会从节点开头重新执行。因此写在 interrupt 前面的数据库写入、扣费、发送消息等副作用必须保证幂等或者移到审批后的独立 Node。关键洞察interrupt 不是“让进程睡眠等待”而是保存 State、结束当前执行等外部输入到来后再按 thread_id 恢复。十、实战用 Python 搭建最小 ReAct 天气 Agent现在用真实 OpenAI API 跑通一个最小案例。用户可以说“西湖那边今天热吗”Agent 需要先把地点解析为标准城市“杭州”再调用天气工具最后返回结构化结果。为了把注意力放在 Agent 机制上天气数据使用本地模拟字典模型调用是真实的 OpenAI API。生产环境只需要把get_weather内部替换成真实天气服务。10.1 安装依赖python -m venv .venvsource .venv/bin/activatepip install -U langchain langgraph langchain-openai pydanticexport OPENAI_API_KEY你的 OpenAI API Key10.2 完整代码import osfrom dataclasses import dataclassfrom langchain.agents import create_agentfrom langchain.chat_models import init_chat_modelfrom langchain.tools import ToolRuntime, toolfrom langgraph.checkpoint.memory import InMemorySaverfrom pydantic import BaseModel, Fielddataclassclass UserContext: user_id: str role: strclass WeatherAnswer(BaseModel): city: str Field(description标准城市名) temperature_c: int Field(description当前摄氏温度) condition: str Field(description天气状况) suggestion: str Field(description一句简短出行建议)tooldef resolve_city(place: str) - str: 把景点、简称或自然语言地点转换成中国标准城市名。 当用户没有直接提供标准城市名时先调用本工具。 aliases { 西湖: 杭州, 魔都: 上海, 羊城: 广州, 帝都: 北京, } return aliases.get(place, place)tooldef get_weather(city: str, runtime: ToolRuntime[UserContext]) - dict: 查询标准城市名对应的当前天气。 只有获得标准城市名后才调用。不要把景点名或城市简称直接传入。 if runtime.context.role not in {member, admin}: raise PermissionError(当前用户没有天气查询权限) weather_data { 杭州: {temperature_c: 31, condition: 多云}, 上海: {temperature_c: 30, condition: 小雨}, 广州: {temperature_c: 33, condition: 晴}, 北京: {temperature_c: 28, condition: 晴}, } return { city: city, **weather_data.get( city, {temperature_c: 25, condition: 暂无实时数据}, ), }model init_chat_model( os.getenv(OPENAI_MODEL, openai:gpt-4.1-mini), temperature0, timeout30, max_retries3,)agent create_agent( modelmodel, tools[resolve_city, get_weather], system_prompt( 你是天气助手。用户输入景点或城市别名时先调用 resolve_city 得到标准城市名后再调用 get_weather最后给出简短出行建议。 ), response_formatWeatherAnswer, context_schemaUserContext, checkpointerInMemorySaver(),)config {configurable: {thread_id: weather-demo-001}}result agent.invoke( { messages: [ {role: user, content: 西湖那边今天热吗} ] }, configconfig, contextUserContext(user_idu-1001, rolemember),)print(result[structured_response])一次典型输出类似city: 杭州temperature_c: 31condition: 多云suggestion: 天气较热建议穿轻薄衣物并注意补水。10.3 这段代码如何形成 ReAct虽然我们没有手写while循环但create_agent已经生成了标准的 Model ↔ Tools 循环第 1 轮 Model 发现“西湖”不是标准城市名 生成 tool_call: resolve_city(place西湖)第 1 次 Tool 返回“杭州” 结果作为 ToolMessage 写回 State第 2 轮 Model 已获得标准城市名 生成 tool_call: get_weather(city杭州)第 2 次 Tool 从 Runtime 读取 role 权限通过后返回结构化天气数据第 3 轮 Model 根据工具结果生成出行建议 按 WeatherAnswer Schema 返回 structured_response在这条链路中代码元素对应组件init_chat_model()Model 统一接口tool 函数ToolsWeatherAnswerresponse formatUserContextruntime contextInMemorySaver()checkpointerthread_id会话状态标识create_agent()组装 Harness 并编译执行图agent.invoke()启动一次运行这就是使用 LangChain 的价值不需要手写 StateGraph也能获得一个标准 ReAct Agent。严格来说这个示例里的resolve_city和get_weather是Tools不是独立 SKILL。Tool 是模型可以直接调用的函数SKILL 更像一套可复用的任务说明、步骤、参考资料和脚本。如果要把它升级成“天气查询 SKILL”可以在 SKILL 中规定城市标准化、数据源选择、异常处理和输出格式再把这两个 Tools 作为执行能力交给 Agent。十一、什么时候应该把它改写成 LangGraph上面的天气 Agent 没有必要直接手写 LangGraph因为它的流程非常标准Model 判断、调用 Tool、拿到结果、继续判断直到输出答案。但如果需求变成下面这样LangGraph 的价值会明显增加复杂天气服务 1. 先判断用户所在租户和数据权限 2. 国内城市走供应商 A海外城市走供应商 B 3. 查询失败时最多重试两次 4. 极端天气进入风险评估节点 5. 高风险预警必须经过人工确认 6. 确认后同时发送短信和企业消息 7. 任一步失败都能从检查点恢复这时应该把关键业务路径显式建模State city、country、weather、risk_level、approved、retry_countNodes resolve_city、route_provider、query_weather、assess_risk human_approval、send_alert、handle_failureEdges 普通边连接固定步骤 条件边根据 country、risk_level、retry_count 选择路径 interrupt 在高风险通知前暂停选择标准不是代码行数而是业务流程是否需要显式控制。能用create_agent middleware清楚表达就先用 LangChain当状态、分支、循环和人工介入成为业务核心再直接设计 LangGraph。十二、生产环境最容易踩的六个坑12.1 Tool description 写得太宽泛“查询信息”几乎没有决策价值。应该说明输入前提、调用时机、能力边界和不能做什么。12.2 把所有控制逻辑塞进 Prompt权限、超时、重试、预算和审计属于确定性控制应该进入 middleware、Tool Gateway 或图节点不应只靠模型自觉遵守。12.3 混淆 context、state 和 storecontext 是本次运行注入的业务依赖state 是流程中不断更新的数据store 是跨会话长期保存的数据。混用会导致权限信息被模型修改或者不同会话互相串数据。12.4 使用 checkpointer 却忘记 thread_idCheckpointer 需要通过thread_id确定保存和恢复哪条会话。首次执行和恢复执行必须使用同一个 ID。12.5 interrupt 前执行不可重复的副作用恢复时节点会从开头重新运行。interrupt 前的扣款、发送消息或数据库写入必须幂等最好拆到审批后的独立节点。12.6 无限制地让 Model ↔ Tools 循环生产 Agent 应限制最大步数、总时长、Token、成本和工具重试次数并监控重复 tool call。否则一个参数错误就可能演变成高成本死循环。写在最后理解 LangChain 和 LangGraph关键不是记住更多 API而是分清两个层次LangChain Model Tools Prompt Middleware Structured Output Runtime Memory 快速搭建一套标准 Agent HarnessLangGraph State Node Edge Branch Loop Interrupt Persistence 精确描述 Agent 如何流转和恢复create_agent帮我们快速得到一套成熟的 Model ↔ Tools 循环middleware 把路由、权限、重试和观测放进执行控制面runtime 把用户、租户和会话信息带入本次运行response format 保证结果能被程序可靠消费。当流程出现复杂分支、长时间运行、人工审批和失败恢复时再把 Agent 展开为 LangGraph让 State 携带数据让 Node 执行步骤让 Edge 决定路径。最终两者不是竞争关系而是同一套 Agent 工程的不同抽象层。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】