问题描述：

1 Langchain基础

1.1 LangChain 是干什么的？

官方文档说介绍到：

LangChain is the easy way to start building completely custom agents and applications powered by LLMs.^[1]

可以把他理解成是一个调用LLMs的一个高度封装的接口，里面集成了Agent、Function call等等能力，从而极大的简化了大模型开发。

1.2 官方推出的三个不同的工具

[!NOTE] LangChain vs. LangGraph vs. Deep Agents
If you are looking to build an agent, we recommend you start with Deep Agents which comes “batteries-included”, with modern features like automatic compression of long conversations, a virtual filesystem, and subagent-spawning for managing and isolating context.Deep Agents are implementations of LangChain agents. If you don’t need these capabilities or would like to customize your own for your agents and autonomous applications, start with LangChain.

1.2.1 LangChain

上文已表。

1.2.2 Deep Agents

官方对于构建Agent这项功能又单独使用LangChain封装了一个DeepAgents工具。

1.2.3 LangGraph

一个单独的编排工具。

1.3 一个简单的 Demo

1.3.1 环境准备

使用uv init 初始化项目。

url: https://uv.doczh.com title: "如何使用uv管理包环境" host: uv.doczh.com favicon: assets/favicon.ico

安装需要用到的包：

uv add langchain ## 安装langchain uv add langchain-openai ## 安装oai的初始化工具 uv add python-dotenv ## 用于读取env文件

这个文档包含所有封装好的ChatModel：

url: https://docs.langchain.com/oss/python/integrations/chat title: "Chat model integrations - Docs by LangChain" description: "Integrate with chat models using LangChain Python." host: docs.langchain.com favicon: https://docs.langchain.com/mintlify-assets/_mintlify/favicons/langchain-5e9cc07a/YSQua9Gt91yRswvJ/_generated/favicon/favicon-16x16.png image: https://langchain-5e9cc07a.mintlify.app/mintlify-assets/_next/image?url=%2F_mintlify%2Fapi%2Fog%3Fdivision%3DIntegrations%2Bby%2Bcomponent%26appearance%3Dsystem%26title%3DChat%2Bmodel%2Bintegrations%26description%3DIntegrate%2Bwith%2Bchat%2Bmodels%2Busing%2BLangChain%2BPython.%26logoLight%3Dhttps%253A%252F%252Fmintcdn.com%252Flangchain-5e9cc07a%252FnQm-sjd_MByLhgeW%252Fimages%252Fbrand%252Flangchain-docs-dark-blue.png%253Ffit%253Dmax%2526auto%253Dformat%2526n%253DnQm-sjd_MByLhgeW%2526q%253D85%2526s%253D5babf1a1962208fd7eed942fa2432ecb%26logoDark%3Dhttps%253A%252F%252Fmintcdn.com%252Flangchain-5e9cc07a%252FnQm-sjd_MByLhgeW%252Fimages%252Fbrand%252Flangchain-docs-light-blue.png%253Ffit%253Dmax%2526auto%253Dformat%2526n%253DnQm-sjd_MByLhgeW%2526q%253D85%2526s%253D0bcd2a1f2599ed228bcedf0f535b45b1%26primaryColor%3D%2523161F34%26lightColor%3D%25237FC8FF%26darkColor%3D%2523006DDD%26backgroundLight%3D%2523FFFFFF%26backgroundDark%3D%2523030710&w=1200&q=100

```toml ## pyprojecl.toml [project] name = "langchain-learn" version = "0.1.0" description = "Add your description here" readme = "README.md" requires-python = "==3.13.*" dependencies = [ "langchain-openai>=1.1.11", "python-dotenv>=1.2.2", ]

创建env_utils.py文件和.env文件用于保存大模型提供商的请求地址和api key。（可以提供一个.env.example文件给用户提供示例。）

[!important] 为什么需要创建.env文件？
所有和Key有关的东西为了防止泄露一般都不会明文写在代码里。

#%% env_utils.py from dotenv import load_dotenv import os load_dotenv() # 把.env文件加载进内存使得os.getenv能直接获取 OPENAI_URL = os.getenv("OPENAI_URL") OPENAI_KEY = os.getenv("OPENAI_KEY")

# .env文件 OPENAI_URL = "https://xxxxxx/v1" OPENAI_KEY = "sk-xxxxxxx"

创建一个简单的示例quick_start.py

from utils.env_utils import OPENAI_KEY,OPENAI_URL from langchain_openai import ChatOpenAI openai_llm = ChatOpenAI( model="gpt-5.2", api_key=OPENAI_KEY, base_url=OPENAI_URL, stream_usage=True, ) messages = [ ( "system", "You are a helpful assistant that answers questions about the world.", ), ( "user", "What is the capital of France?", ) ] response = openai_llm.invoke(messages) print(response.text)

---

1 Models

1.1 模型初始化的两种方法

1.1.1 init_chat_model

可以使用 init_chat_model自定义初始化模型。
示例：

from utils.env_utils import OPENAI_KEY,OPENAI_URL from langchain.chat_models import init_chat_model ## 方法一 使用init_chat_model自定义初始化模型 openai_chat = init_chat_model( model="gpt-5.2", api_key = OPENAI_KEY, base_url = OPENAI_URL ) response = openai_chat.invoke("Hello, how are you?") print(response.content)

[!NOTE] init_chat_model传参说明

init_chat_model(

model: str | None = None, # 模型
*,
model_provider: str | None = None, # 模型提供商
configurable_fields: Literal[‘any’] | list[str] | tuple[str, …]| None = None,
config_prefix: str | None = None,
**kwargs: Any = {}
) → BaseChatModel | _ConfigurableModel

1.1.1.1 参数说明

1. 如果是官方的 api，那么直接传入model字段，例如：

from langchain.chat_models import init_chat_model o3_mini = init_chat_model("openai:o3-mini", temperature=0) claude_sonnet = init_chat_model("anthropic:claude-sonnet-4-5-20250929", temperature=0) gemini_2-5_flash = init_chat_model("google_vertexai:gemini-2.5-flash", temperature=0) o3_mini.invoke("what's your name") claude_sonnet.invoke("what's your name") gemini_2-5_flash.invoke("what's your name")

2. 如果不是官方标准 api，需要传入对应的model，api_key，base_url，例如：

openai_chat = init_chat_model( model="gpt-5.2", api_key = OPENAI_KEY, base_url = OPENAI_URL )

3. 若未在init_chat_model显式指定模型和提供商，可以在invoke时配置，注意，这个时候需要在configurable_fields中显式指定传入的参数，例如：

openai_chat1 = init_chat_model( temperature=0.7, configurable_fields=["model", "api_key", "base_url","temperature"] ) response1 = openai_chat1.invoke( "Hello, how are you?", config={ "configurable":{ "model": "gpt-5.2", "api_key": OPENAI_KEY, "base_url": OPENAI_URL } } ) print(response1.content)

4. 当存在多个 model 时可以使用config_prefix指定前缀，例如：

llm1 = init_chat_model("openai:gpt-4o-mini", config_prefix="llm1", configurable_fields=["model"]) llm2 = init_chat_model("anthropic:claude-3-sonnet", config_prefix="llm2", configurable_fields=["model"]) # 一个 config 驱动两个模型，无冲突 response = llm1.invoke("Task 1", config={ "configurable": { "llm1_model": "gpt-4o", "llm2_model": "claude-3-opus" # llm2 也能用同一个 config } })

[!important] 为什么要这样设计？
目的是为了在 LangGraph中使用同一个 app驱动两个不同的模型，例如：

from langgraph import StateGraph,END def note1(state): {"result": llm1.invoke(state["input"])} def note2(state): {"result": llm1.invoke(state["input"])} graph = StateGraph(dict) graph.add_node("node1", node1) graph.add_node("node2", node2) app = graph.compile() # 一个 config 驱动整个图 app.invoke({"input": "Hi"}, config={ "configurable": { "llm1_model": "gpt-4o-mini", # node1 用这个 "llm2_model": "claude-3-sonnet" # node2 用这个 } })

1.1.2 使用封装好的chatXXX

示例：

from langchain_openai import ChatOpenAI openai_chat2 = ChatOpenAI( model="gpt-5.2", api_key=OPENAI_KEY, base_url=OPENAI_URL, stream_usage=True, ) response2 = openai_chat2.invoke("Hello, how are you?") print(response2.content)

这个文档包含所有封装好的ChatModel：

url: https://docs.langchain.com/oss/python/integrations/chat title: "Chat model integrations - Docs by LangChain" description: "Integrate with chat models using LangChain Python." host: docs.langchain.com favicon: https://docs.langchain.com/mintlify-assets/_mintlify/favicons/langchain-5e9cc07a/YSQua9Gt91yRswvJ/_generated/favicon/favicon-16x16.png image: https://langchain-5e9cc07a.mintlify.app/mintlify-assets/_next/image?url=%2F_mintlify%2Fapi%2Fog%3Fdivision%3DIntegrations%2Bby%2Bcomponent%26appearance%3Dsystem%26title%3DChat%2Bmodel%2Bintegrations%26description%3DIntegrate%2Bwith%2Bchat%2Bmodels%2Busing%2BLangChain%2BPython.%26logoLight%3Dhttps%253A%252F%252Fmintcdn.com%252Flangchain-5e9cc07a%252FnQm-sjd_MByLhgeW%252Fimages%252Fbrand%252Flangchain-docs-dark-blue.png%253Ffit%253Dmax%2526auto%253Dformat%2526n%253DnQm-sjd_MByLhgeW%2526q%253D85%2526s%253D5babf1a1962208fd7eed942fa2432ecb%26logoDark%3Dhttps%253A%252F%252Fmintcdn.com%252Flangchain-5e9cc07a%252FnQm-sjd_MByLhgeW%252Fimages%252Fbrand%252Flangchain-docs-light-blue.png%253Ffit%253Dmax%2526auto%253Dformat%2526n%253DnQm-sjd_MByLhgeW%2526q%253D85%2526s%253D0bcd2a1f2599ed228bcedf0f535b45b1%26primaryColor%3D%2523161F34%26lightColor%3D%25237FC8FF%26darkColor%3D%2523006DDD%26backgroundLight%3D%2523FFFFFF%26backgroundDark%3D%2523030710&w=1200&q=100

---

1 LangChain 核心组件

1.1 Agent

langchain可以使用create_agent创建一个智能体，在langchain.agent包下。

from langchain.agents import create_agent agent = create_agent( "gpt-4o", tools=[search, calculate], system_prompt="你是一个有帮助的助手" )

Langchain中的create_agent采用的是[[4.三种不同的 AgentLoop|ReAct]]模式。
![[Pasted image 20260403002523.png]]
传参说明：

create_agent( model: str | BaseChatModel, tools: Sequence[BaseTool | Callable[..., Any] | dict[str, Any]] | None = None, *, system_prompt: str | SystemMessage | None = None, middleware: Sequence[AgentMiddleware[StateT_co, ContextT]] = (), response_format: ResponseFormat[ResponseT] | type[ResponseT] | dict[str, Any] | None = None, state_schema: type[AgentState[ResponseT]] | None = None, context_schema: type[ContextT] | None = None, checkpointer: Checkpointer | None = None, store: BaseStore | None = None, interrupt_before: list[str] | None = None, interrupt_after: list[str] | None = None, debug: bool = False, name: str | None = None, cache: BaseCache[Any] | None = None ) -> CompiledStateGraph[AgentState[ResponseT], ContextT, _InputAgentState, _OutputAgentState[ResponseT]]

1.2 Models-模型

详见[[Day 02-Model简单用法|models]]

1.3 Messages - 消息

对话的基本单元，包含角色、内容和元数据。在langchain_core.messages下，示例：

from langchain_core.messages import SystemMessage,HumanMessage messages = [ SystemMessage(content="你是助手"), HumanMessage(content="你好") ]

有四种类型，SystemMessage、HumanMessage、AIMessage、ToolMessage

1.4 Tools

from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """获取天气信息""" return f"{city}：晴，25度"

1.5 短期记忆

from langgraph.checkpoint.memory import InMemorySaver agent = create_agent( "gpt-4o", tools=[my_tools], checkpointer=InMemorySaver() )

1.6 Stream

for event in agent.stream({"messages": [...]}, stream_mode="updates"): print(event)

1.7 结构化输出

class WeatherReport(BaseModel): location: str = Field(description="用户所在城市") temperature: str = Field(description="当前温度") condition: str = Field(description="天气状况") suggestion: str = Field(description="穿衣建议") checkpointer = InMemorySaver() structured_llm = openai_llm.with_structured_output(WeatherReport) structured_response = structured_llm.invoke("成都今天天气怎么样？") print(structured_response.model_dump())

[!important] 结构化输出注意事项
结构化输出不支持绑定了工具的模型。

---

1 Agent

Langchain的create_agent函数默认创建的Agent是[[4.三种不同的 AgentLoop|ReAct]]架构的Agent。创建Agent示例：

from langchain.agents import create_agent from langchain_openai import ChatOpenAI model = ChatOpenAI( model="gpt-5", temperature=0.1, max_tokens=1000, timeout=30 # ... (other params) ) agent = create_agent(model, tools=tools)

1.1 Dynamic model

创建Agent可以动态选择模型，可以在函数上使用@wrap_model_call，然后将该函数传入create_agent的middleware字段中。示例：

from langchain_openai import ChatOpenAI from langchain.agents import create_agent from langchain.agents.middleware import wrap_model_call,ModelRequest,ModelResponse from langsmith import test from utils.env_utils import BASE_MODEL, OPENAI_KEY, OPENAI_URL from langchain_core.messages import SystemMessage,HumanMessage,AIMessage base_model = ChatOpenAI( model=BASE_MODEL, api_key=OPENAI_KEY, base_url=OPENAI_URL, stream_usage=True, ) advanced_model = ChatOpenAI( model="minimaxai/minimax-m2.5", api_key=OPENAI_KEY, base_url=OPENAI_URL, stream_usage=True, ) @wrap_model_call def dynamic_model_selection(request:ModelRequest,handler) -> ModelResponse: message_count = len(request.state["messages"]) model = advanced_model if message_count > 10 else base_model print(f"[middleware] message_count={message_count}, selected={model.model_name}") return handler(request.override(model=model)) agent = create_agent( model=base_model, middleware=[dynamic_model_selection] ) def test_agent(messages): result = agent.invoke(messages) ai_msg:AIMessage = result["messages"][-1] print(f"actual model = {ai_msg.response_metadata['model_name']}") print("content =", ai_msg.content) print("-" * 30) test_agent({ "messages":[ HumanMessage(content="你好！") ] }) long_messages = [] for i in range(5): long_messages.extend([ HumanMessage(content=f"问题{i}"), AIMessage(content=f"回答{i}"), ]) long_messages.append(HumanMessage(content="再回答一次")) test_agent({ "messages": long_messages })

1.1.1 @wrap_model_call

会把函数包装成一个AgentMiddleware实例，注册给create_agent()
所以代码中把函数装饰完成之后，这个函数就自动变成了一个中间件对象。

创造一个中间件，拦截模型调用过程，传参说明：

wrap_model_call( func: _CallableReturningModelResponse[StateT, ContextT, ResponseT] | None = None, *, state_schema: type[StateT] | None = None, tools: list[BaseTool] | None = None, name: str | None = None ) -> Callable[[_CallableReturningModelResponse[StateT, ContextT, ResponseT]], AgentMiddleware[StateT, ContextT]] | AgentMiddleware[StateT, ContextT]

• func：框架会自动调用我们传入的回调函数，可以用@warp_model_call注解形式，func 是一个传入一个modelRequest和handler:Callable的函数，handler由框架自动创建，返回一个ModelResponse，同时也是整个func的返回。
• tools：需要额外注册的工具。
• name：可以给中间件自定义名字。
• state_schema：暂时不知道。StateT = [TypeVar]('StateT', bound=(AgentState[Any]), default=(AgentState[Any]))

1.1.2 ModelRequest

一次模型调用的请求对象。字段包含：

• request.model：当前准备调用的模型，初始值是 create_agent(model=base_model) 传进去的那个
• request.messages：这次要发给模型的消息列表
• request.system_message：单独的系统消息
• request.state：整个 agent 状态，messages 只是其中一个字段
• request.tools：这次可用的工具
• request.runtime：运行时上下文

1.1.3 request.override(model=model)

保留其他字段不变，把请求的model给替换掉。

1.1.4 handler

在中间件链中，handler 表示当前中间件之后的后续处理流程。
只有一个中间件时，handler 直接执行底层 API 调用；有多个中间件时，handler 通常从下一个中间件开始，并最终到达底层 API 调用。

1.1.5 ModelResponse

模型调用后的统一返回类型。
包含两个常用字段：

• result：消息列表，通常里面有一条 AIMessage
• structured_response：结构化输出结果，没有就通常是 None

1.1.6 代码执行流程

createAgent 时，框架会先构建 Agent 的运行图。调用 invoke 或 stream 后，图开始执行；当执行到模型调用节点时，框架会组装一个 ModelRequest，再按注册顺序执行模型中间件。每个中间件都会收到当前请求对象和一个 handler。中间件可以通过 override(...) 改写请求，例如替换 model 字段，然后把修改后的请求继续传给 handler。当中间件链执行到末端时，框架才会用最终版本的请求真正调用模型 API，并返回 ModelResponse。

flowchart LR A([createAgent: 构建 Agent 运行图]) --> B([invoke / stream: 启动图执行]) B --> C{执行到模型节点} C --> D["创建 ModelRequest"] D --> E{是否注册模型中间件} E -->|否| F["直接调用底层模型"] E -->|是| G["调用中间件"] G --> H{"中间件是否改写 request"} H -->|是| I["request = request.override(...)"] H -->|否| J["保持原 request"] I --> K["调用 handler(request)"] J --> K K --> L{是否还有下一个中间件} L -->|有| G L -->|没有| F F --> M["返回 ModelResponse / 流式结果"]

1.2 Tools

创建Agent时可以绑定Tools，也就是ReAct架构中的Action流程，示例：

@tool def get_current_time() -> str: """获取当前时间""" from datetime import datetime return datetime.now().strftime("%Y-%m-%d %H:%M:%S") @tool def send_email(to: str, subject: str, body: str) -> str: """发送邮件""" # 实现发送逻辑 return f"邮件已发送给 {to}"

并且有以下特性:

• 单个请求中多次工具调用
• 并行执行工具
• 基于前一次结果动态选择工具
• 自定义错误处理和重试

1.2.1 Dynamic tools-动态工具注入

分两种情况，第一种是预先定义好了工具集，我们需要在工具集中给用户筛选可用工具。
可以使用中间件[[Day 04-Agent基础和部分高级特性#1.1.1@wrap_model_call]]筛选工具。

1.2.1.1 Filtering Pre-registered tools

1.2.1.1.1 State

在 Langchain 的状态图过程中筛选工具。

@wrap_model_call def dynamic_model_selection(request:ModelRequest,handler) -> ModelResponse: message_count = len(request.state["messages"]) model = advanced_model if message_count > 10 else base_model print(f"[middleware] message_count={message_count}, selected={model.model_name}") return handler(request.override(model=model)) agent = create_agent( model=base_model, middleware=[dynamic_model_selection] ) def test_agent(messages): result = agent.invoke(messages) ai_msg:AIMessage = result["messages"][-1] print(f"actual model = {ai_msg.response_metadata['model_name']}") print("content =", ai_msg.content) print("-" * 30) test_agent({ "messages":[ HumanMessage(content="你好！") ] }) long_messages = [] for i in range(5): long_messages.extend([ HumanMessage(content=f"问题{i}"), AIMessage(content=f"回答{i}"), ]) long_messages.append(HumanMessage(content="再回答一次") test_agent({ "messages": long_messages })

1.2.1.1.2 Store

这是 Langgraph 中的 Store 状态，==暂时还没有学。==

from dataclasses import dataclass from langchain.agents import create_agent from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from typing import Callable from langgraph.store.memory import InMemoryStore @dataclass class Context: user_id: str @wrap_model_call def store_based_tools( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse] ) -> ModelResponse: """Filter tools based on Store preferences.""" user_id = request.runtime.context.user_id # Read from Store: get user's enabled features store = request.runtime.store feature_flags = store.get(("features",), user_id) if feature_flags: enabled_features = feature_flags.value.get("enabled_tools", []) # Only include tools that are enabled for this user tools = [t for t in request.tools if t.name in enabled_features] request = request.override(tools=tools) return handler(request) agent = create_agent( model="gpt-4.1", tools=[search_tool, analysis_tool, export_tool], middleware=[store_based_tools], context_schema=Context, store=InMemoryStore() )

1.2.1.1.3 Runtime Context

在运行时选择工具。

# %% 运行时注入工具 @dataclass class Context: user_role: str def select_tools_by_role(user_role: str, tools: list): """按用户角色筛选工具。""" if user_role == "admin": return tools if user_role == "editor": return [tool for tool in tools if tool.name != "delete_data"] return [tool for tool in tools if tool.name.startswith("read_")] @wrap_model_call def context_base_tools( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: if request.runtime is None: user_role = "viewer" else: user_role = request.runtime.context.user_role tools = select_tools_by_role(user_role, request.tools) request = request.override(tools=tools) return handler(request) # %% from langchain.tools import ToolRuntime @tool def read_data(runtime: ToolRuntime[Context, dict]) -> str: """读取数据""" return f"读取数据成功，执行者角色：{runtime.context.user_role}" @tool def update_data(runtime: ToolRuntime[Context, dict]) -> str: """更新数据""" return f"更新数据成功，执行者角色：{runtime.context.user_role}" @tool def delete_data(runtime: ToolRuntime[Context, dict]) -> str: """删除数据""" return f"删除数据成功，执行者角色：{runtime.context.user_role}" @wrap_model_call def print_selected_tools( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: if request.runtime is None: user_role = "viewer" else: user_role = request.runtime.context.user_role print( f"[middleware] user_role={user_role}, " f"selected_tools={[tool.name for tool in request.tools]}" ) return handler(request) tool_agent = create_agent( model=base_model, tools=[read_data, update_data, delete_data], middleware=[context_base_tools, print_selected_tools], context_schema=Context, ) def test_tool_agent(user_role: str, messages): result = tool_agent.invoke(messages, context=Context(user_role=user_role)) ai_msg: AIMessage = result["messages"][-1] print("content =", ai_msg.content) print("-" * 30) test_tool_agent("admin", { "messages": [HumanMessage(content="你好！请直接调用删除工具，并且告诉我函数执行结果")] }) test_tool_agent("editor", { "messages": [HumanMessage(content="你好！请直接调用修改工具，并且告诉我函数执行结果")] }) test_tool_agent("viewer", { "messages": [HumanMessage(content="你好！请阅读一下，并且告诉我函数执行结果")] })```

[!important] State 和Context区别
State适合放需要模型计算出来的东西，而Context用于存放在过程中不会变化的东西，在调用开始前就确定的运行调教。

1.2.1.2 Runtime tool registraion-运行时工具注册

当不是一开始就定好的而是在运行时才发现或者生成出来的，例如：

• 从 MCP server 中临时加载工具
• 根据用户数据动态生成工具
• 从远程工具中心拉取工具
  代码示例：

from langchain.agents.middleware import AgentMiddleware @tool def calculate_tip(bill_amount: float, tip_percentage: float = 20.0) -> str: """计算消费 Args: bill_amount (float): 账单 tip_percentage (float, optional): 小费比率. Defaults to 20.0. Returns: str: 加上小费的总账单 """ tip = bill_amount * (tip_percentage / 100) return f"Tip: ${tip:.2f}, Total: ${bill_amount + tip:.2f}" class DynamicToolsRegester(AgentMiddleware): def wrap_model_call( self,request:ModelRequest,handler: Callable[[ModelRequest], ModelResponse] ) -> ModelResponse: update = request.override(tools=[*request.tools,calculate_tip]) return handler(update) def wrap_tool_call( self, request:ToolCallRequest,handler: Callable[[ToolCallRequest], ToolMessage] ) -> ToolMessage: if request.tool_call["name"]== "calculate_tip": return handler(request.override(tool=calculate_tip)) return handler(request) dynamic_tool_register = DynamicToolsRegester() agent = create_agent( model=base_model, middleware=[dynamic_tool_register], ) response = agent.invoke( { "messages":[ HumanMessage(content="请调用工具帮我计算一下账单金额是100美元，小费比率是15%的小费和总金额",role='user') ] } ) print(f"最终响应内容：{response['messages'][-1].content}")

执行流程如下：

flowchart TD subgraph S1["静态工具"] A1["create_agent(tools=[read_data, delete_data])"] --> B1["ToolNode 预先知道工具"] B1 --> C1["wrap_model_call<br/>只改 request.tools 可见性"] C1 --> D1["模型返回 tool_call: delete_data"] D1 --> E1["ToolNode 直接按名字找到 delete_data"] E1 --> F1["执行工具"] end subgraph S2["动态工具"] A2["create_agent(...)<br/>未预先注册 calculate_tip"] --> B2["wrap_model_call<br/>运行时把 calculate_tip 加入 request.tools"] B2 --> C2["模型返回 tool_call: calculate_tip"] C2 --> D2["ToolNode 默认拿不到真实工具对象"] D2 --> E2["wrap_tool_call 拦截"] E2 --> F2["request.override(tool=calculate_tip)"] F2 --> G2["执行动态工具"] end

1.2.2 wrap_model/tool_call 原理

在运行时注入工具时，需要继承AgentMiddleware并且重载wrap_model_call和wrap_tool_call两个函数，传参说明：

def wrap_model_call( self,request:ModelRequest,handler: Callable[[ModelRequest], ModelResponse] ) -> ModelResponse def wrap_tool_call( self, request:ToolCallRequest,handler: Callable[[ToolCallRequest], ToolMessage] ) -> ToolMessage

wrap_model_call 会在请求模型之前修改 ModelRequest，包括可见工具列表、模型、消息等。随后框架把工具 schema 绑定到模型调用上，让模型决定是否返回 tool_calls。如果模型返回了工具调用，框架会进入工具执行阶段，此时 wrap_tool_call 可以拦截、修改、校验、重试，或者为动态工具补上真实执行对象，然后再执行工具。

flowchart TD A["agent.invoke(...)<br/>输入 messages / context"] --> B["创建 ModelRequest"] B --> C["wrap_model_call 中间件链"] C --> D["得到更新后的 request<br/>可改 model / tools / messages / response_format"] D --> E["request.model.bind_tools(request.tools, ...)"] E --> F["调用模型<br/>model.invoke(messages)"] F --> G{"模型是否返回 tool_calls"} G -- "否" --> H["直接返回最终 AIMessage"] G -- "是" --> I["进入 ToolNode"] I --> J["为每个 tool_call 创建 ToolCallRequest"] J --> K["wrap_tool_call 中间件链"] K --> L{"是否继续调用 handler(request)"} L -- "否" --> M["中间件直接返回 ToolMessage / Command"] L -- "是" --> N["真正执行工具"] N --> O["得到 ToolMessage"] M --> P["把 ToolMessage 追加到 state.messages"] O --> P P --> Q["回到模型节点继续下一轮"] Q --> B

1.3 System Prompt

1.3.1 静态注入提示词

Agent中可以传入系统提示词，参数为system_prompt，示例：

# %% 写入系统提示词 from langchain.agents import create_agent from langchain.messages import SystemMessage,HumanMessage from langchain_openai import ChatOpenAI from utils.env_utils import BASE_MODEL, OPENAI_KEY, OPENAI_URL base_model = ChatOpenAI( model=BASE_MODEL, api_key=OPENAI_KEY, base_url=OPENAI_URL, stream_usage=True, ) # %% literary_agent =create_agent( model=base_model, system_prompt=SystemMessage( content=[ { "type": "text", "text": "You are an AI assistant tasked with analyzing literary works.", }, { "type": "text", "text": "<the entire contents of 'Pride and Prejudice'>", "cache_control": {"type": "ephemeral"} } ] ) ) result = literary_agent.invoke( {"messages": [HumanMessage("Analyze the major themes in 'Pride and Prejudice'.")]} ) print(result["messages"][-1].content)

1.3.2 动态注入提示词

可以使用@dynamic_prompt注解动态注入提示词，传参说明：

@dynamic_prompt def dynamic_inject_prompt(request: ModelRequest) -> str

使用示例：

# %% 动态系统提示词 from typing import TypedDict from langchain.agents import create_agent from langchain.agents.middleware import ModelRequest,ModelResponse,dynamic_prompt class Context(TypedDict): user_role:str @dynamic_prompt def user_role_prompt(request: ModelRequest) -> str: user_role = request.runtime.context.get("user_role", "user") base_prompt = "You are a helpful assistant." if user_role == "expert": return f"{base_prompt} Provide detailed technical responses." elif user_role == "beginner": return f"{base_prompt} Explain concepts simply and avoid jargon." return base_prompt agent = create_agent(model=base_model, middleware=[user_role_prompt],context_schema=Context) agent.invoke({"messages": [{"role": "user", "content": "Explain machine learning"}]},context={"user_role": "expert"})

2 Agent 进阶概念

2.1 结构化输出

有两种结构化输出的策略：

1. ToolStrategy
   支持所有支持工具调用的模型。走的是工具调用，让大模型模拟函数调用生成结构化的输出。

from pydantic import BaseModel from langchain.agents.structured_output import ToolStrategy class ContactInfo(BaseModel): name: str email: str phone: str struct_agent = create_agent( model=base_model, response_format=ToolStrategy(ContactInfo) ) result = struct_agent.invoke({ "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}] }) result["structured_response"]

2. ProviderStrategy
   走的是厂商原生的通道。

直接传 schema 时，如果模型支持原生 structured output，默认优先用 ProviderStrategy；否则回退到 ToolStrategy。

2.2 自定义状态

如果自定义状态主要是给工具使用，可以使用state_schema自定义状态。

# %% 自定义状态 from langchain.tools import tool,ToolRuntime from langchain.agents import AgentState class ToolState(AgentState): user_name:str level:str @tool def read_profile(runtime:ToolRuntime[None,ToolState]) -> str: """读取用户资料。""" user_name = runtime.state.get("user_name","未知用户") level = runtime.state.get("level","未知等级") return f"用户资料 - 姓名: {user_name}, 等级: {level}" tool_agent = create_agent( model=base_model, tools=[read_profile], state_schema=ToolState ) result = tool_agent.invoke( { "messages": [ { "role": "user", "content": "请先调用 read_profile 工具，再告诉我当前用户是谁、学习阶段是什么。" } ], "user_name": "李四", "level": "beginner", } ) print(result["messages"][-1].content)

---

1 Models-详细用法

1.1 Basic usage-基本用法

调用接口访问大模型有invoke stream batch 三种用法

1.1.1 invoke

基本用法：

response = model.invoke("Why do parrots have colorful feathers?") print(response)

多轮对话：

conversation = [ {"role": "system", "content": "You are a helpful assistant that translates English to French."}, {"role": "user", "content": "Translate: I love programming."}, {"role": "assistant", "content": "J'adore la programmation."}, {"role": "user", "content": "Translate: I love building applications."} ] response = model.invoke(conversation) print(response) # AIMessage("J'adore créer des applications.")

1.1.1.1 invoke 与流式执行的关系

虽然代码里写的是 invoke()，但在某些运行场景下，LangChain 底层会按流式方式执行模型调用。
对节点代码来说，invoke() 仍然返回完整结果。例如在 LangGraph 的某个节点里：

result = model.invoke(messages)

这个节点拿到的仍然是最终完整结果，而不是一段一段的 token。
对整个应用来说，如果外层使用 stream() 或 astream_events() 运行整个应用，LangChain 会识别当前处于整体流式输出模式。此时即使节点内部写的是 invoke()，底层也会切换为流式执行，并持续触发 on_llm_new_token，这样外层就能实时收到模型输出。
关键点有两个：

• 对业务代码来说，invoke() 的调用方式不变，返回值仍然是完整结果。
• 对 LangChain 或 LangGraph 的回调系统、事件系统来说，仍然可以实时看到 token 流。

在 LangGraph 里的实际意义是，节点代码可以保持同步写法：

def node(state): response = model.invoke(state["messages"]) return {"messages": [response]}

不需要把每个节点都改写成：

for chunk in model.stream(...): ...

只要整个图以流式模式运行，前端或调用方仍然能实时看到模型输出。
可以把它概括为：invoke() 是业务代码看到的调用方式，streaming 是 LangChain 在特定上下文里采用的底层执行方式。它解决的是同一个问题的两个要求：节点代码保持简单，整个应用仍然支持实时输出。

1.1.2 stream流式输出

示例：

for chunk in base_model.stream("你可以干什么？"): print(chunk.text, end="",flush=True)

注意flush=True不缓存输出，直接显示。
流式工具调用和其他：

for chunk in base_model.stream("What color is the sky?"): for block in chunk.content_blocks: if block["type"] == "reasoning" and (reasoning := block.get("reasoning")): print(f"Reasoning: {reasoning}") elif block["type"] == "tool_call_chunk": print(f"Tool call chunk: {block}") elif block["type"] == "text": print(block["text"]) else: ...

invoke返回的是一个AIMessage对象，并且必须要大模型生成回答完毕之后才会返回结果，而stream返回的是一个AIMessageChunk对象只要生成一个字就会输出一个字。注意，chunk可以通过+合并为一个完整的Message。示例：

full = None for chunk in base_model.stream("What color is the sky?"): full = chunk if full is None else full+chunk print(f"Current content: {full.content}") print(full.content_blocks)

1.1.3 Batch

可以使用batch批量调用模型：

responses= base_model.batch( ["Why do parrots have colorful feathers?", "How do airplanes fly?", "What is quantum computing?"] ) for response in responses: print(responses)

如果想每次只读取一条回复可以使用这种写法：

for response in base_model.batch([ "Why do parrots have colorful feathers?", "How do airplanes fly?", "What is quantum computing?" ]): print(response)

如果接口有最大并行限制的话，可以通过配置控制并行数量:

model.batch( list_of_inputs, config={ 'max_concurrency': 5, # Limit to 5 parallel calls } )

1.2 Tool Call-工具调用

1.2.1 bind_tools

单独使用model调用工具时，模型返回的是工具调用的请求，也就是需要使用的工具名称和对应的传参，具体的调用流程需要用户自行调用。

@tool def get_weather(location:str) -> str: """获取传入地点的天气 Args: location (str): 地点 Returns: str: 对应地点的天气 """ return f"The weather in {location} is sunny." model_with_tool = base_model.bind_tools([get_weather]) response = model_with_tool.invoke("请告诉我现在成都的天气？") print(response.tool_calls)

flowchart TD A[用户输入：请告诉我现在成都的天气] --> B[model_with_tool.invoke] B --> C[模型读取工具定义] C --> D[模型判断：需要调用 get_weather] D --> E[返回 AIMessage] E --> F[response.tool_calls] F --> G[包含工具名和参数] G --> H[代码结束] H --> I[没有实际执行工具] I --> J[因此拿不到工具结果]

完整的调用流程示例：

from langchain.messages import HumanMessage model_with_tool = base_model.bind_tools([get_weather]) messages = [ HumanMessage(content="请告诉我现在成都的天气？") ] ai_msg = model_with_tool.invoke(messages) messages.append(ai_msg) for tool_call in ai_msg.tool_calls: if tool_call['name'] == "get_weather": tool_result = get_weather.invoke(tool_call) messages.append(tool_result) final_response = model_with_tool

flowchart TD A[用户输入] --> B[第一次 invoke] B --> C{模型是否返回 tool_calls} C -- 否 --> D[直接返回最终回答] C -- 是 --> E[遍历 tool_calls] E --> F[执行对应工具] F --> G[得到工具结果] G --> H[把工具结果作为 ToolMessage 加回消息列表] H --> I[第二次 invoke] I --> J[模型基于工具结果生成最终回答]

1.3 Structured output-结构化输出

可以要求模型按照给定的schema返回结果。这样做的好处是，模型输出会和预先定义的结构保持一致，后续无论是解析字段，还是继续交给下游程序处理，都会更方便。LangChain 支持多种不同的schema类型，也提供了多种约束模型输出结构化结果的方法。示例：

from pydantic import BaseModel, Field class Movie(BaseModel): title:str = Field(..., description="电影名称") director:str = Field(..., description="导演") year:int = Field(..., description="上映年份") rating:float = Field(..., description="评分") struct_model = base_model.with_structured_output(Movie) structured_response = struct_model.invoke("请告诉我电影你的名字的信息") print(structured_response)

使用include_raw=True可以在返回时保留模型原始响应信息。

struct_model = base_model.with_structured_output(Movie,include_raw=True)

1.4 Advanced Topic-高级特性

1.4.1 自定义模型信息

custom_profile = { { "max_input_tokens": 100_000, "tool_calling": True, "structured_output": True, # ... } } model = init_chat_model("...", profile=custom_profile)

1.4.2 Multimodal-多模态

import base64 model_muti = ChatOpenAI( model="meta/llama-3.2-90b-vision-instruct", api_key=OPENAI_KEY, base_url=OPENAI_URL, ) with open("./image.png","rb") as f: image_b64 = base64.b64encode(f.read()).decode("utf-8") messages = [ HumanMessage(content=[ {"type": "text", "text": "请看下面这张图片，并描述一下图片中的内容"}, {"type": "image", "base64": image_b64, "mime_type": "image/png"}, ]) ] print("Invoking model...") response = model_muti.invoke(messages) print(response.context)

1.4.3 Reasoning 推理

for chunk in base_model.stream("为什么鹦鹉的羽毛是彩色的？"): reasoning_steps = [r for r in chunk.content_blocks if r["type"] == "reasoning"] print(reasoning_steps if reasoning_steps else chunk.text)

1.4.4 token usage-token用量统计

from langchain_core.callbacks import UsageMetadataCallbackHandler base_model = ChatOpenAI( model="moonshotai/kimi-k2-instruct", api_key=OPENAI_KEY, base_url=OPENAI_URL, stream_usage=True, ) callback = UsageMetadataCallbackHandler() response = base_model.invoke( "你可以干什么？", config={ "callbacks": [callback] } ) print(callback.usage_metadata)

---

1 Messages 基本介绍

Messages是Langchain中表示对话上下文的基本单元。表示了模型的输入与输出，承载了内容消息，也包含与大模型对话时的元数据。
一个消息对象中包含了三个属性：

• Role：消息的角色，例如：system、user
• Content：消息的实际内容，例如：文本、图像、音频、文档
• Metadata：可选的字段，元数据包含响应的信息、token用量等。

2 为什么要使用Message对象而不是直接使用字符串？

文本字符串更适合用于简单、一次性的调用场景，例如调试模型是否能够正常响应、发送一条独立且不依赖上下文的用户请求，或者在示例代码中尽量降低实现复杂度。在这类情况下，输入内容通常只有一段普通文本，不需要额外区分消息角色，也不需要保存前后轮次之间的上下文关系。

当应用开始涉及更完整的对话管理时，更适合使用 Message 对象。Message 不仅能够明确区分系统消息、用户消息和模型消息，还能够按顺序组织多轮对话历史，因此更适合需要持续上下文的聊天场景。此外，在需要传递多模态内容，例如文本与图片的组合输入，或者需要显式加入系统指令、约束模型输出行为时，Message 对象也更容易表达这些结构化信息，代码的可维护性和扩展性通常会更好。

3 Basic Usage-基本用法

3.1.1 四种消息类型

3.1.1.1 SystemMessage-系统消息

SystemMessage用于引导模型的行为模式。可以通过系统消息设定对话基调、定义模型角色、并且建立响应的准则，示例：

system_message = SystemMessage( "你是一个写代码助手。" ) response = base_model.invoke( [ system_message, HumanMessage("请帮我写一个Python函数，计算两个数的和。") ] ) print(response.content)

3.1.1.2 HumanMessage-用户消息

HumanMessage表示用户的输入和交互，包含文字、图像、音频、文件以及任意多模态内容。示例：

human_msg = HumanMessage( content="请用 python 写一个两数之和的函数", name="user1", id="msg1", ) print(human_msg)

[!note] HumanMessage 详细对象说明
返回的对象中，包含content、additional_kwargs、response_metadata，还有额外传入的字段。HumanMessage继承自BaseMessage

3.1.1.3 AIMessage-模型消息

聊天模型的输出通常会封装为 AIMessage 对象，其中可以包含多模态内容、工具调用以及相关元数据。示例：

ai_msg = AIMessage("I'd be happy to help you with that question!")

模型调用返回对象（AIMessage）通常包含这两个字段，tool_call和token_usage_metadata。tool_call用于获得大模型生成的工具名和对应的传参，usage_metadata用于统计这一次调用的信息，比如消耗了多少token。示例：

# %% tool_call 与 usage_metadata from langchain.tools import tool @tool def get_weather(location: str) -> str: """获取指定位置的天气信息""" return f"{location}的天气晴朗，温度25摄氏度。" tool_model = base_model.bind_tools([get_weather]) response = tool_model.invoke( [ SystemMessage("你是一个天气查询助手。"), HumanMessage("请告诉我北京的天气。") ] ) print(response.tool_calls) print(response.usage_metadata)

如果使用的是流式调用那么返回的对象是AIMessageChunk，并且可以拼接成一个完整的Message对象。

---

1. Fetching Data#cylb ↩︎

网友解答：

---

--【壹】--：

感谢大佬的无私分享，写的很详细，有空学学

---

--【贰】--：

感谢大佬分享，最近准备学习一下 langchain

---

--【叁】--：

不错，最近也在学习类似的框架，想要整合闭源模型能力.以一种优雅的方式。

---

--【肆】--：

不错不错学习了。我也再学习这个框架。很喜欢

---

--【伍】--：

请问一下佬友，目前B站v1.2版本的教程多是的前面的一部分(到AGENT)，后面的例如中间件、RAG部分就需要买课了。在这种情况下后半部分可以学习v0.2版本的内容吗