先讲清楚 1.0 改了什么如果你在网上看到大量LLMChain、AgentExecutor、initialize_agent的教程那些大多是 0.x 时代的写法。LangChain 1.0 做了一次不小的收敛理解这次变化比记 API 更重要。2.1 包结构从“大杂烩”到分层langchain-core 抽象层Runnable、消息、工具、回调、输出解析 langchain 高层create_agent、内置中间件、模型初始化 langchain-classic 旧版LLMChain、AgentExecutor、旧 Memory 等兼容用 langchain-openai 供应商集成Anthropic、Google 等各自独立包 langgraph 底层运行时状态图、检查点、持久化 langchain-text-splitters 文本切分几个直接后果依赖变轻了不再为了用一个模型拖进整个生态langchain里只剩一个高层 Agent 抽象旧的 Chain/Agent 执行器移到了langchain-classicLangGraph 从“LangChain 的一个子项目”变成真正的运行时底座create_agent编译出来就是一个图。2.2 一个高层抽象create_agent1.0 的核心判断是大多数 Agent 需求都可以归结为“模型 工具 循环 中间件”。于是官方只保留一个入口from langchain.agents import create_agent agent create_agent( modelopenai:gpt-4.1-mini, tools[...], system_prompt你是一个严谨的助手。, )它内部生成的是 LangGraph 图调用模型 → 如果有 tool calls 就执行工具 → 把结果回填 → 再调用模型直到没有工具调用或达到停止条件。当你需要更复杂的控制分支、并行、人工审批、子图不必换框架直接下沉到 LangGraph 的StateGraph。两者共享同一套消息、工具、回调生态。2.3 标准内容块一套消息格式适配所有模型模型输出早就不只是字符串了推理模型的 thinking、引用、服务端工具调用、图片、音频。0.x 时代每个 provider 各有一套content结构业务代码到处写if provider ...。1.0 引入了标准内容块standard content blocksmessage.content # 原始格式可能是 str也可能是 provider 原生块列表 message.content_blocks # 统一后的标准块列表 message.text # 纯文本快捷方式常见块类型text、reasoning、image、tool_call、tool_result、citation。这让上层代码可以只处理标准块把 provider 差异留在集成包里。2. 模型初始化把“换模型”变成改一行2.1init_chat_modelfrom langchain.chat_models import init_chat_model model init_chat_model( openai:gpt-4.1-mini, # provider:model temperature0.2, timeout60, max_retries2, )字符串前缀决定 providerinit_chat_model(anthropic:claude-sonnet-4-5) init_chat_model(google_genai:gemini-2.5-pro) init_chat_model(openai:deepseek-chat, base_urlhttps://api.deepseek.com/v1) init_chat_model(ollama:qwen3:8b)init_chat_model的价值在于把模型选择变成配置项# config.py chat_model: str openai:gpt-4.1-minimodel init_chat_model(settings.chat_model, temperature0.2)测试、灰度、降级都可以通过改配置完成而不是改代码。2.2 直接用集成包需要 provider 特有参数时直接用对应类from langchain_openai import ChatOpenAI model ChatOpenAI( modelgpt-4.1-mini, temperature0, timeout60, max_retries2, stream_usageTrue, # 流式时也返回 token 用量 )建议业务代码依赖BaseChatModel接口invoke/ainvoke/stream/astream/bind_tools不要在业务层散落具体类。这样换模型只需要改一个工厂函数。2.3 关键参数怎么设参数建议temperature抽取、分类、工具调用用 0~0.2创意写作用 0.7timeout单次请求超时必须小于网关超时max_retries只对 429/5xx/网络错误有意义不要对参数错误重试max_tokens显式设置防止失控输出注意它限制的是输出不是上下文stream_usage流式场景需要统计 token 时打开3. 消息LLM 应用的“数据结构”3.1 四种基本角色from langchain.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ) messages [ SystemMessage(你是一个只回答技术问题的助手。), HumanMessage(什么是幂等性), AIMessage(幂等性是指同一操作执行多次结果与执行一次相同。), HumanMessage(给一个 HTTP 的例子。), ]不要用{role: user, content: ...}裸字典。消息对象有tool_calls、usage_metadata、response_metadata等字段裸字典会在工具调用、追踪、多模态场景丢信息。3.2 多模态与标准内容块from langchain.messages import HumanMessage msg HumanMessage(content_blocks[ {type: text, text: 这张图里有什么}, {type: image, url: https://example.com/chart.png}, ])content_blocks是标准格式如果某个 provider 还不支持某个块集成层会报错或降级而不是静默丢失。3.3 工具调用消息模型发起工具调用时AIMessage.tool_calls是一个列表[ { name: get_weather, args: {city: 杭州}, id: call_abc123, type: tool_call, } ]执行完工具后必须用同一个id构造ToolMessage回填ToolMessage(content杭州 26℃多云, tool_call_idcall_abc123)漏掉tool_call_id或顺序错乱是工具调用 400 错误最常见的原因。4. 工具让模型“能做事”4.1 定义工具from typing import Literal from pydantic import BaseModel, Field from langchain.tools import tool class WeatherInput(BaseModel): city: str Field(description城市名例如杭州) unit: Literal[celsius, fahrenheit] celsius tool(get_weather, args_schemaWeatherInput) def get_weather(city: str, unit: str celsius) - str: 查询指定城市的当前天气。 当用户询问天气、温度、是否下雨时使用。 # 实际项目里这里调用天气 API return f{city} 当前 26°C多云工具定义的三条铁律docstring 就是给模型看的说明书。写清楚“什么时候用、参数含义、返回什么”比在系统提示里重复描述有效得多参数用 Pydantic 约束。枚举、范围、必填/可选能显著降低模型传错参数的概率返回字符串要短且结构化。工具结果会进入上下文返回 10KB 的原始 JSON 是 token 浪费。4.2 手动跑一遍工具循环理解 Agent 的关键很多人直接用create_agent却不知道里面发生了什么。手写一次后面调试会轻松很多from langchain.messages import HumanMessage, ToolMessage TOOLS {get_weather: get_weather} model_with_tools model.bind_tools([get_weather]) messages [HumanMessage(杭州今天天气怎么样)] response model_with_tools.invoke(messages) messages.append(response) while response.tool_calls: for call in response.tool_calls: tool_fn TOOLS[call[name]] try: result tool_fn.invoke(call[args]) except Exception as exc: result f工具执行失败{exc} messages.append(ToolMessage(result, tool_call_idcall[id])) response model_with_tools.invoke(messages) messages.append(response) print(response.text)这个循环就是 Agent 的心脏。create_agent帮你加上了状态持久化、并行工具调用、错误重试、最大轮数、中间件、流式事件。4.3 工具错误返回错误还是抛异常可预期的业务错误城市不存在、订单号格式错返回一段错误说明让模型自己纠正或向用户解释不可恢复的错误鉴权失败、下游宕机抛异常交给重试/降级/告警不要让异常 traceback 直接进上下文既浪费 token 又可能泄露内部信息。LangChain 的ToolExceptionhandle_tool_error可以把异常转成工具消息适合前一类场景。4.4 工具设计的高级原则少而正交5 个语义清晰工具好过 20 个功能重叠的工具。工具太多会显著降低选择准确率。工具内部做鉴权不信任模型参数模型可以传任意user_id必须用服务端会话里的身份覆盖。幂等模型可能重复调用同一个工具。写操作要有幂等键。可观测工具名、参数摘要、耗时、成功/失败都要进 trace敏感参数要脱敏。控制返回体积检索类工具返回 top-k 摘要分页参数显式暴露不要把整个数据库塞回去。5. 结构化输出把“自然语言”变成“可编程数据”5.1with_structured_outputfrom typing import Literal from pydantic import BaseModel, Field class Ticket(BaseModel): 从用户描述中抽取的工单信息。 title: str Field(description不超过 30 字的标题) priority: Literal[P0, P1, P2, P3] Field(description紧急程度) category: Literal[billing, bug, account, other] tags: list[str] Field(default_factorylist, max_length5) structured_model model.with_structured_output(Ticket, methodfunction_calling) ticket structured_model.invoke(线上支付全挂了客户无法下单已经持续 20 分钟) print(ticket.priority) # P0method的选择方法原理适用function_calling把 schema 作为工具让模型调用大多数支持工具调用的模型兼容性最好json_schema用 provider 的结构化输出能力支持 strict JSON schema 的模型约束更强json_mode要求模型输出 JSON再解析不支持工具调用的模型稳定性较差5.2 校验与重试结构化输出不是 100% 可靠。生产代码必须假设它会失败from pydantic import ValidationError def extract_ticket(text: str, retries: int 2) - Ticket: last_error None for _ in range(retries 1): try: return structured_model.invoke(text) except (ValidationError, ValueError) as exc: last_error exc # 可以把错误反馈给模型再试一次不要无限重试 raise RuntimeError(f结构化输出失败{last_error})更优雅的做法是用 Agent 的response_formatToolStrategy它内置了“模型返回多个结果/格式错误”时的纠错回环from langchain.agents import create_agent from langchain.agents.structured_output import ToolStrategy agent create_agent( modelopenai:gpt-4.1-mini, tools[], response_formatToolStrategy(Ticket), ) result agent.invoke({messages: [{role: user, content: 支付挂了P0}]}) print(result[structured_response])5.3 设计 schema 的经验字段越少越稳。先问“业务真正需要哪几个字段”再谈扩展用Literal代替自由字符串做分类给每个字段写description这是模型理解字段含义的主要依据日期、金额用明确格式YYYY-MM-DD、以分为单位的整数不要让模型猜嵌套结构超过两层考虑拆成多次抽取。6.create_agent一个入口四种扩展点6.1 基本用法from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver agent create_agent( modelopenai:gpt-4.1-mini, tools[get_weather, search_orders], system_prompt( 你是客服助手。回答必须基于工具结果不要编造订单信息。 涉及退款、改地址等写操作时先向用户确认。 ), checkpointerInMemorySaver(), ) config {configurable: {thread_id: user-42}} result agent.invoke( {messages: [{role: user, content: 我的订单到哪了}]}, config, ) print(result[messages][-1].text)6.2 内部循环与停止条件create_agent生成的图大致是START → model → (有 tool_calls?) → tools → model → ... → END停止条件模型不再发起工具调用达到recursion_limit默认 25 步左右可按 graph 配置调整中间件主动jump_toend触发中断human-in-the-loop。别把recursion_limit调得很大来“解决”死循环。死循环通常是工具返回没有进展、提示词缺少停止条件或者模型在重复同一个错误调用。先修逻辑再谈上限。6.3 中间件1.0 最值得学的部分中间件是“在 Agent 循环的关键位置插入横切逻辑”的机制。常见钩子钩子时机典型用途before_model每次调用模型前注入上下文、裁剪历史、检查预算、动态改提示词after_model模型返回后校验输出、记录用量、触发人工审核wrap_model_call包裹模型调用重试、降级、缓存、超时wrap_tool_call包裹工具调用鉴权、限流、审计、错误转换dynamic_prompt构建提示词时按用户/租户/步骤切换提示词内置中间件from langchain.agents import create_agent from langchain.agents.middleware import ( SummarizationMiddleware, HumanInTheLoopMiddleware, PIIMiddleware, ) from langgraph.checkpoint.memory import InMemorySaver agent create_agent( modelopenai:gpt-4.1-mini, tools[get_weather, send_email], middleware[ # 输入中的邮箱自动脱敏 PIIMiddleware(email, strategyredact), # 上下文超过 8000 token 时自动摘要保留最近 12 条 SummarizationMiddleware( modelopenai:gpt-4.1-mini, trigger(tokens, 8000), keep(messages, 12), ), # 发送邮件前暂停等待人工批准 HumanInTheLoopMiddleware(interrupt_on{send_email: True}), ], checkpointerInMemorySaver(), # HITL 必须有 checkpointer )内置中间件覆盖了生产中最常见的几类需求Summarization长对话压缩控制上下文成本HumanInTheLoop敏感工具调用前暂停PII邮箱、信用卡、身份证等脱敏/屏蔽Model/Tool Retry上游抖动时自动重试Call Limit限制模型/工具调用次数防失控Model Fallback主模型失败时切备用模型。中间件的具体类名和参数在小版本间可能调整。写代码前以当前官方文档为准模式不变import 路径可能变。自定义中间件装饰器写法最轻from typing import Any, Callable from langchain.agents import create_agent from langchain.agents.middleware import ( AgentState, ModelRequest, ModelResponse, before_model, wrap_model_call, ) from langgraph.runtime import Runtime before_model def log_context(state: AgentState, runtime: Runtime) - dict[str, Any] | None: print(f调用模型前消息数{len(state[messages])}) return None wrap_model_call def retry_model( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) - ModelResponse: for attempt in range(3): try: return handler(request) except Exception: if attempt 2: raise raise RuntimeError(unreachable) agent create_agent( modelopenai:gpt-4.1-mini, tools[get_weather], middleware[log_context, retry_model], )更复杂的状态扩展用类from typing_extensions import NotRequired from langchain.agents.middleware import AgentState class SupportState(AgentState): user_id: NotRequired[str] step: NotRequired[str] class BudgetMiddleware: def before_model(self, state: SupportState, runtime): if state.get(step) done: return {jump_to: end} return None中间件的价值在于把“每次调用都要做的事”从业务提示词里拿出来变成可测试、可复用的代码。审计、脱敏、预算、重试都不应该靠模型自觉。6.4 记忆短期靠 checkpointer长期靠 store短期记忆同一会话checkpointerthread_id。消息历史、工具结果、状态都按 thread 持久化进程重启不丢。长期记忆跨会话LangGraph 的store按namespace存用户偏好、事实、画像。它不是消息历史而是“应该被记住的知识”。from langgraph.checkpoint.memory import InMemorySaver agent create_agent(..., checkpointerInMemorySaver()) config {configurable: {thread_id: user-42:chat-7}}生产环境把InMemorySaver换成PostgresSaver/SqliteSaver。内存版只适合开发和测试进程一重启就没了。6.5 流式async for chunk in agent.astream( {messages: [{role: user, content: 介绍一下杭州}]}, stream_modemessages, ): message_chunk, metadata chunk if message_chunk.text: print(message_chunk.text, end, flushTrue)stream_mode常用值messages模型 token 流最适合聊天updates每个节点执行后的状态增量values完整状态快照custom节点内通过get_stream_writer()写出的自定义事件进度、工具状态。需要统一的追踪事件时用astream_eventsasync for event in agent.astream_events(inputs, versionv2): if event[event] on_chat_model_stream: ... elif event[event] on_tool_start: ...7. RAG检索增强生成不是“接个向量库”RAG 的完整链路动图交互版 可切换明暗主题。离线文档 → 解析 → 清洗 → 切分 → 向量化 → 索引 元数据 在线问题 → 查询改写 → 检索 → 重排 → 上下文组装 → 生成 → 引用7.1 最小实现from langchain_core.documents import Document from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough from langchain_openai import OpenAIEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.vectorstores import InMemoryVectorStore raw_docs [Document(page_contenttext, metadata{source: handbook.md})] splitter RecursiveCharacterTextSplitter( chunk_size800, chunk_overlap120, separators[\n\n, \n, 。, , , , ], ) chunks splitter.split_documents(raw_docs) vectorstore InMemoryVectorStore.from_documents( chunks, OpenAIEmbeddings(modeltext-embedding-3-small), ) retriever vectorstore.as_retriever( search_typemmr, search_kwargs{k: 6, fetch_k: 20}, ) def format_docs(docs: list[Document]) - str: return \n\n.join( f[{i1}] {d.page_content}\n来源{d.metadata.get(source)} for i, d in enumerate(docs) ) prompt ChatPromptTemplate.from_messages([ (system, 只根据给定资料回答。资料中没有的信息明确说“资料未提及”。 回答末尾列出引用的资料编号。), (human, 资料\n{context}\n\n问题{question}), ]) chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | model | StrOutputParser() ) answer chain.invoke(年假怎么计算)7.2 决定效果的不是向量库是这几件事环节关键决策常见问题解析PDF 表格、扫描件、HTML 噪音解析丢结构切分后语义断裂切分按语义/标题切而不是固定字符chunk 太大稀释相关性太小丢上下文元数据来源、章节、时间、权限无法过滤召回无关内容查询改写、多查询、HyDE用户口语与文档术语不匹配检索向量 BM25 混合、MMR纯向量对专有名词、编号不敏感重排cross-encoder / LLM reranktop-k 里相关文档排不到前面上下文去重、压缩、按 token 预算裁剪塞太多导致“lost in the middle”生成强制引用、拒答策略模型用参数知识编造7.3 进阶模式Hybrid SearchEnsembleRetriever合并 BM25 与向量检索权重可调Multi-Query让模型生成 3~5 个改写问题分别检索后合并Parent Document用小 chunk 检索返回大 chunk 给模型Self-Query从问题中抽取元数据过滤条件时间、部门、文档类型Contextual Retrieval给每个 chunk 加上“它属于哪篇文档的哪一节”的上下文再向量化Rerank先用向量召回 50 条再用交叉编码器精排到 6 条。一个常被忽略的点RAG 的评估要分开看“检索质量”和“生成质量”。检索用 recallk、MRR生成用忠实度、引用准确率、答案相关性。混在一起看永远不知道是该换 embedding 还是改提示词。第 04、05 篇会讲怎么用 LangSmith / Langfuse 做这件事。8. 回调与可观测性LangChain 的 callback 系统是追踪的基础。所有invoke/stream都接受configfrom langfuse.langchain import CallbackHandler handler CallbackHandler() result agent.invoke( {messages: [...]}, config{callbacks: [handler], run_name: support-chat}, )LangSmith 更简单环境变量打开即可LANGSMITH_TRACINGtrue LANGSMITH_API_KEYyour-key LANGSMITH_PROJECTllm-app-dev一次 Agent 调用会被拆成树状 tracechain → model → tool → retriever。你能看到每一步的输入输出、token、延迟、错误。第 04、05 篇会展开讲。9. 什么时候不该用 LangChain这不是“反框架”而是工程判断只有一个模型调用、没有工具/检索直接用官方 SDK少一层抽象需要极致性能或 provider 独有能力直接调 SDK避免抽象层丢参数团队已有成熟的内部抽象别为了用而用LangChain 的迁移成本不低复杂状态机、强一致流程直接用 LangGraph甚至用普通代码 状态机别硬套 Agent。LangChain 真正省事的地方多 provider 统一接口工具调用、结构化输出、流式、回调这些“每次都要写”的胶水大量文档加载器、向量库、检索器集成与 LangSmith/Langfuse 的原生追踪。10. 本篇检查清单知道langchain、langchain-core、langchain-classic、langgraph的分工模型通过工厂/配置初始化业务层不散落 provider 类用消息对象不用裸字典工具 docstring 写清使用时机参数有 Pydantic 约束能手写一遍 tool-call 循环理解tool_call_id结构化输出有校验和有限重试长对话用了摘要/裁剪而不是无限堆历史敏感工具调用有人工审批中间件RAG 的检索与生成分开评估知道什么时候该下沉到 LangGraph下一篇03 | LangGraph 1.x有状态 Agent 的工程化。create_agent能覆盖大部分场景但一旦你需要“暂停等待审批”“重启后接着跑”“多 Agent 分工”“按状态走不同分支”就需要直接操作状态图。那才是 Agent 工程真正开始的地方。
