先讲清楚 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-classic；

• LangGraph 从“LangChain 的一个子项目”变成真正的运行时底座，create_agent编译出来就是一个图。

2.2 一个高层抽象：create_agent

1.0 的核心判断是：大多数 Agent 需求都可以归结为“模型 + 工具 + 循环 + 中间件”。于是官方只保留一个入口：

from langchain.agents import create_agent ​ agent = create_agent( model="openai: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 blocks）：

message.content # 原始格式，可能是 str，也可能是 provider 原生块列表 message.content_blocks # 统一后的标准块列表 message.text # 纯文本快捷方式

常见块类型：text、reasoning、image、tool_call、tool_result、citation。这让上层代码可以只处理标准块，把 provider 差异留在集成包里。

2. 模型初始化：把“换模型”变成改一行

2.1init_chat_model

from langchain.chat_models import init_chat_model ​ model = init_chat_model( "openai:gpt-4.1-mini", # provider:model temperature=0.2, timeout=60, max_retries=2, )

字符串前缀决定 provider：

init_chat_model("anthropic:claude-sonnet-4-5") init_chat_model("google_genai:gemini-2.5-pro") init_chat_model("openai:deepseek-chat", base_url="https://api.deepseek.com/v1") init_chat_model("ollama:qwen3:8b")

init_chat_model的价值在于把模型选择变成配置项：

# config.py chat_model: str = "openai:gpt-4.1-mini"

model = init_chat_model(settings.chat_model, temperature=0.2)

测试、灰度、降级都可以通过改配置完成，而不是改代码。

2.2 直接用集成包

需要 provider 特有参数时，直接用对应类：

from langchain_openai import ChatOpenAI ​ model = ChatOpenAI( model="gpt-4.1-mini", temperature=0, timeout=60, max_retries=2, stream_usage=True, # 流式时也返回 token 用量 )

建议：业务代码依赖BaseChatModel接口（invoke/ainvoke/stream/astream/bind_tools），不要在业务层散落具体类。这样换模型只需要改一个工厂函数。

2.3 关键参数怎么设

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_id="call_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_schema=WeatherInput) def get_weather(city: str, unit: str = "celsius") -> str: """查询指定城市的当前天气。 当用户询问天气、温度、是否下雨时使用。 """ # 实际项目里这里调用天气 API return f"{city} 当前 26°C，多云"

工具定义的三条铁律：

1. docstring 就是给模型看的说明书。写清楚“什么时候用、参数含义、返回什么”，比在系统提示里重复描述有效得多；

2. 参数用 Pydantic 约束。枚举、范围、必填/可选，能显著降低模型传错参数的概率；

3. 返回字符串要短且结构化。工具结果会进入上下文，返回 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_id=call["id"])) response = model_with_tools.invoke(messages) messages.append(response) print(response.text)

这个循环就是 Agent 的心脏。create_agent帮你加上了：状态持久化、并行工具调用、错误重试、最大轮数、中间件、流式事件。

4.3 工具错误：返回错误还是抛异常？

• 可预期的业务错误（城市不存在、订单号格式错）：返回一段错误说明，让模型自己纠正或向用户解释；

• 不可恢复的错误（鉴权失败、下游宕机）：抛异常，交给重试/降级/告警；

• 不要让异常 traceback 直接进上下文，既浪费 token 又可能泄露内部信息。

LangChain 的ToolException+handle_tool_error可以把异常转成工具消息，适合前一类场景。

4.4 工具设计的高级原则

• 少而正交：5 个语义清晰工具，好过 20 个功能重叠的工具。工具太多会显著降低选择准确率。

• 工具内部做鉴权，不信任模型参数：模型可以传任意user_id，必须用服务端会话里的身份覆盖。

• 幂等：模型可能重复调用同一个工具。写操作要有幂等键。

• 可观测：工具名、参数摘要、耗时、成功/失败都要进 trace；敏感参数要脱敏。

• 控制返回体积：检索类工具返回 top-k 摘要，分页参数显式暴露，不要把整个数据库塞回去。

5. 结构化输出：把“自然语言”变成“可编程数据”

5.1with_structured_output

from 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_factory=list, max_length=5) structured_model = model.with_structured_output(Ticket, method="function_calling") ticket = structured_model.invoke("线上支付全挂了，客户无法下单，已经持续 20 分钟") print(ticket.priority) # P0

method的选择：

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_format+ToolStrategy，它内置了“模型返回多个结果/格式错误”时的纠错回环：

from langchain.agents import create_agent from langchain.agents.structured_output import ToolStrategy agent = create_agent( model="openai:gpt-4.1-mini", tools=[], response_format=ToolStrategy(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( model="openai:gpt-4.1-mini", tools=[get_weather, search_orders], system_prompt=( "你是客服助手。回答必须基于工具结果，不要编造订单信息。" "涉及退款、改地址等写操作时，先向用户确认。" ), checkpointer=InMemorySaver(), ) 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_to="end"；

• 触发中断（human-in-the-loop）。

别把recursion_limit调得很大来“解决”死循环。死循环通常是工具返回没有进展、提示词缺少停止条件，或者模型在重复同一个错误调用。先修逻辑，再谈上限。

6.3 中间件：1.0 最值得学的部分

中间件是“在 Agent 循环的关键位置插入横切逻辑”的机制。常见钩子：

内置中间件

from langchain.agents import create_agent from langchain.agents.middleware import ( SummarizationMiddleware, HumanInTheLoopMiddleware, PIIMiddleware, ) from langgraph.checkpoint.memory import InMemorySaver agent = create_agent( model="openai:gpt-4.1-mini", tools=[get_weather, send_email], middleware=[ # 输入中的邮箱自动脱敏 PIIMiddleware("email", strategy="redact"), # 上下文超过 8000 token 时自动摘要，保留最近 12 条 SummarizationMiddleware( model="openai:gpt-4.1-mini", trigger=("tokens", 8000), keep=("messages", 12), ), # 发送邮件前暂停，等待人工批准 HumanInTheLoopMiddleware(interrupt_on={"send_email": True}), ], checkpointer=InMemorySaver(), # 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( model="openai: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

• 短期记忆（同一会话）：checkpointer+thread_id。消息历史、工具结果、状态都按 thread 持久化，进程重启不丢。

• 长期记忆（跨会话）：LangGraph 的store，按namespace存用户偏好、事实、画像。它不是消息历史，而是“应该被记住的知识”。

from langgraph.checkpoint.memory import InMemorySaver agent = create_agent(..., checkpointer=InMemorySaver()) config = {"configurable": {"thread_id": "user-42:chat-7"}}

生产环境把InMemorySaver换成PostgresSaver/SqliteSaver。内存版只适合开发和测试，进程一重启就没了。

6.5 流式

async for chunk in agent.astream( {"messages": [{"role": "user", "content": "介绍一下杭州"}]}, stream_mode="messages", ): message_chunk, metadata = chunk if message_chunk.text: print(message_chunk.text, end="", flush=True)

stream_mode常用值：

• messages：模型 token 流，最适合聊天；

• updates：每个节点执行后的状态增量；

• values：完整状态快照；

• custom：节点内通过get_stream_writer()写出的自定义事件（进度、工具状态）。

需要统一的追踪事件时用astream_events：

async for event in agent.astream_events(inputs, version="v2"): 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_content=text, metadata={"source": "handbook.md"})] splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=120, separators=["\n\n", "\n", "。", "！", "？", " ", ""], ) chunks = splitter.split_documents(raw_docs) vectorstore = InMemoryVectorStore.from_documents( chunks, OpenAIEmbeddings(model="text-embedding-3-small"), ) retriever = vectorstore.as_retriever( search_type="mmr", search_kwargs={"k": 6, "fetch_k": 20}, ) def format_docs(docs: list[Document]) -> str: return "\n\n".join( f"[{i+1}] {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 决定效果的不是向量库，是这几件事

7.3 进阶模式

• Hybrid Search：EnsembleRetriever合并 BM25 与向量检索，权重可调；

• Multi-Query：让模型生成 3~5 个改写问题，分别检索后合并；

• Parent Document：用小 chunk 检索，返回大 chunk 给模型；

• Self-Query：从问题中抽取元数据过滤条件（时间、部门、文档类型）；

• Contextual Retrieval：给每个 chunk 加上“它属于哪篇文档的哪一节”的上下文再向量化；

• Rerank：先用向量召回 50 条，再用交叉编码器精排到 6 条。

一个常被忽略的点：RAG 的评估要分开看“检索质量”和“生成质量”。检索用 recall@k、MRR；生成用忠实度、引用准确率、答案相关性。混在一起看，永远不知道是该换 embedding 还是改提示词。第 04、05 篇会讲怎么用 LangSmith / Langfuse 做这件事。

8. 回调与可观测性

LangChain 的 callback 系统是追踪的基础。所有invoke/stream都接受config：

from langfuse.langchain import CallbackHandler handler = CallbackHandler() result = agent.invoke( {"messages": [...]}, config={"callbacks": [handler], "run_name": "support-chat"}, )

LangSmith 更简单，环境变量打开即可：

LANGSMITH_TRACING=true LANGSMITH_API_KEY=<your-key> LANGSMITH_PROJECT=llm-app-dev

一次 Agent 调用会被拆成树状 trace：chain → 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 工程真正开始的地方。