Back to Blog

LangChain使用

聊天模型接收字符串、消息列表或 PromptValue，返回。初始化模型不等于调用模型；初始化只得到一个可复用的模型对象。统一入口：也可以使用供应商专用类，例如、或。常用参数： | 参数 | 含义 | | ------------- | --------------------------------...

Type AI

Created Jun 23, 2026, 21:19:00 / Updated Jun 23, 2026, 21:19:00

Views — / Comments — / Words 4,268

模型创建与调用

聊天模型接收字符串、消息列表或 PromptValue，返回 AIMessage。初始化模型不等于调用模型；初始化只得到一个可复用的模型对象。

创建模型

统一入口：

from langchain.chat_models import init_chat_model

model = init_chat_model(
    model="provider:model-name",
    temperature=0.2,
    timeout=30,
    max_retries=2,
)

也可以使用供应商专用类，例如 ChatOpenAI、ChatDeepSeek 或 ChatOllama。常用参数：

参数	含义
`model`	模型名称，统一入口可同时包含 Provider
`temperature`	随机性；抽取、分类和代码任务通常用较低值
`max_tokens`	最大输出 Token，部分 Provider 使用其他名称
`timeout`	单次请求超时
`max_retries`	SDK 层请求重试次数
`api_key`	API Key，通常从环境变量读取
`base_url`	OpenAI 兼容服务或中转服务地址

返回值是聊天模型对象，它实现 Runnable 接口，可继续调用 invoke、stream、batch、ainvoke 等方法。

`invoke()`

from langchain.messages import HumanMessage

response = model.invoke([
    HumanMessage(content="用一句话介绍 LangChain")
])

输入形式：

输入	示例
字符串	`model.invoke("你好")`
消息列表	`model.invoke([SystemMessage(...), HumanMessage(...)])`
PromptValue	`model.invoke(prompt.invoke(values))`

返回 AIMessage。关键字段：

字段	类型	含义
`content`	字符串或内容块列表	模型正文；工具调用时可能为空
`id`	`Optional[str]`	LangChain 运行消息 ID
`usage_metadata`	`Optional[dict]`	统一 Token 统计
`response_metadata`	`dict`	Provider 元数据，如模型名和结束原因
`tool_calls`	`list[dict]`	已解析的工具调用请求
`invalid_tool_calls`	`list`	解析失败的工具调用

print(response.content)
print(response.id)
print(response.usage_metadata)
print(response.response_metadata.get("finish_reason"))

usage_metadata 常含 input_tokens、output_tokens 和 total_tokens。字段可能为 None，明细取决于 Provider。

`stream()`

for chunk in model.stream("写一段产品介绍"):
    print(chunk.content, end="")

返回 AIMessageChunk 迭代器。每个 chunk 可能包含增量 content、工具调用分块和响应元数据。合并完整消息：

full = None
for chunk in model.stream("你好"):
    full = chunk if full is None else full + chunk

print(full.content)

流式 Token 用量是否出现取决于模型及配置，不应假设每个 chunk 都有用量。

`batch()` 与 `batch_as_completed()`

responses = model.batch(["问题一", "问题二"])

batch() 返回 list[AIMessage]，顺序与输入一致。

for index, response in model.batch_as_completed(["问题一", "问题二"]):
    print(index, response.content)

batch_as_completed() 按完成顺序产生 (原输入索引, AIMessage)，适合各请求耗时差异较大的场景。

异步方法

方法	返回值
`await model.ainvoke(input)`	`AIMessage`
`model.astream(input)`	`AIMessageChunk` 异步迭代器
`await model.abatch(inputs)`	`list[AIMessage]`
`model.abatch_as_completed(inputs)`	`(index, AIMessage)` 异步迭代器

`config` 参数

Runnable 调用可传入运行配置：

response = model.invoke(
    "你好",
    config={
        "tags": ["customer-service"],
        "metadata": {"user_id": "u-1"},
        "run_name": "answer-question",
    },
)

config 用于追踪、回调、并发限制和运行元数据，不会自动成为提示词内容。

消息与提示词

消息类型

类型	来源	用途
`SystemMessage`	应用	角色、规则和系统约束
`HumanMessage`	用户输入层	用户消息
`AIMessage`	聊天模型	模型正文或工具调用请求
`ToolMessage`	工具执行层	工具调用结果
`RemoveMessage`	状态管理层	从消息状态中删除指定消息

from langchain.messages import SystemMessage, HumanMessage

messages = [
    SystemMessage(content="你是 Python 助手"),
    HumanMessage(content="解释生成器"),
]
response = model.invoke(messages)

消息的公共字段包括 content、id、name、additional_kwargs 和 response_metadata。AIMessage 另有 usage_metadata、tool_calls、invalid_tool_calls；ToolMessage 另有 tool_call_id。

content 可能是字符串，也可能是 Provider 原生内容块。需要统一内容块时读取 content_blocks，但不同模型支持程度不同。

`ChatPromptTemplate`

提示词模板负责把变量渲染成消息列表。

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{domain}领域助手"),
    ("human", "请回答：{question}"),
])

prompt_value = prompt.invoke({
    "domain": "Python",
    "question": "什么是迭代器？",
})

prompt.invoke() 返回 ChatPromptValue，常用方法：

方法	返回值
`prompt_value.to_messages()`	渲染后的消息对象列表
`prompt_value.to_string()`	拼接后的字符串表示

模板不调用模型。它的输出可以交给模型：

response = model.invoke(prompt_value)

也可组成 Runnable 管道：

chain = prompt | model
response = chain.invoke({"domain": "Python", "question": "解释迭代器"})

此时返回值由管道最后一个组件决定，因此仍是 AIMessage。

消息占位符

from langchain_core.prompts import MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是客服助手"),
    MessagesPlaceholder("history", optional=True),
    ("human", "{question}"),
])

value = prompt.invoke({
    "history": previous_messages,
    "question": "刚才说到哪里？",
})

history 必须是消息列表。占位符用于插入多轮历史、工具消息或其他已存在的消息对象。

`partial()`

python_prompt = prompt.partial(domain="Python")
response = (python_prompt | model).invoke({"question": "什么是装饰器？"})

partial() 返回新的模板对象，提前固定部分变量，不会执行模板或模型。

结构化输出

with_structured_output(schema) 为模型绑定输出 Schema。它返回新的 Runnable；调用该 Runnable 时，结果会被解析成 Schema 对应的 Python 对象。

Pydantic

from pydantic import BaseModel, Field

class Person(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")

structured_model = model.with_structured_output(Person)
person = structured_model.invoke("张三今年 30 岁")

print(person.name)
print(person.age)

返回值是 Person 实例。Pydantic 会执行运行时校验，类型不满足时可能抛出验证或解析异常。

TypedDict、JSON Schema 与 dataclass

课件中的四种 Schema 返回差异：

Schema	`invoke()` 返回值	运行时校验
Pydantic Model	Pydantic 实例	有
`TypedDict`	`dict`	通常无 Pydantic 式校验
JSON Schema	`dict`	依赖模型和调用策略
`dataclass`	`dict`	通常不自动实例化 dataclass

使用 TypedDict：

from typing_extensions import TypedDict, Annotated

class PersonDict(TypedDict):
    name: Annotated[str, "姓名"]
    age: Annotated[int, "年龄"]

structured_model = model.with_structured_output(PersonDict)
result = structured_model.invoke("张三今年 30 岁")
print(result["name"])

保留原始消息

structured_model = model.with_structured_output(Person, include_raw=True)
result = structured_model.invoke("张三今年 30 岁")

返回字典通常包含：

字段	含义
`raw`	原始 `AIMessage`，可读取 Token 和响应元数据
`parsed`	解析后的 Pydantic 实例或字典
`parsing_error`	解析异常；成功时通常为 `None`

只使用解析结果时，不开启 include_raw；需要 Token、原始工具调用或排查解析问题时开启。

与 Agent 的区别

模型结构化输出：

structured_model = model.with_structured_output(Person)
result = structured_model.invoke(input_text)

直接返回 Person 或字典。

Agent 结构化输出通过 response_format 配置，agent.invoke() 仍返回 Agent State 字典，结构化结果位于 result["structured_response"]。详见 Agent 文档。

工具

工具是带名称、描述和参数 Schema 的可调用对象。模型只负责生成工具调用请求，真正执行工具的是应用或 Agent 运行时。

定义工具

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。"""
    return f"{city}：晴，25°C"

装饰后，get_weather 不再只是普通函数，而是 Tool 对象：

属性	含义
`.name`	暴露给模型的工具名
`.description`	模型选择工具时读取的描述
`.args_schema`	参数 Schema
`.args`	JSON Schema 形式的参数说明

直接调用：

result = get_weather.invoke({"city": "北京"})

返回值是工具函数自身的结果，本例为字符串。

自定义参数 Schema

from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")

@tool(args_schema=WeatherInput)
def get_weather(city: str) -> str:
    """查询天气。"""
    return f"{city}：晴"

字段描述会进入工具 Schema，帮助模型正确生成参数。工具函数签名应和 args_schema 保持一致。

绑定模型

model_with_tools = model.bind_tools([get_weather])
ai_message = model_with_tools.invoke("北京天气如何？")

bind_tools() 返回新的模型 Runnable，不执行工具。invoke() 返回 AIMessage，工具请求位于：

ai_message.tool_calls == [
    {
        "name": "get_weather",
        "args": {"city": "北京"},
        "id": "call_123",
        "type": "tool_call",
    }
]

若模型选择直接回答，tool_calls 为空，正文在 content。

手动执行完整流程

from langchain.messages import HumanMessage, ToolMessage

messages = [HumanMessage(content="北京天气如何？")]
ai_message = model_with_tools.invoke(messages)
messages.append(ai_message)

for call in ai_message.tool_calls:
    result = get_weather.invoke(call["args"])
    messages.append(ToolMessage(
        content=str(result),
        name=call["name"],
        tool_call_id=call["id"],
    ))

final_message = model_with_tools.invoke(messages)
print(final_message.content)

ToolMessage.tool_call_id 必须与请求的 id 对应，否则模型无法确认结果属于哪次调用。

`tool_choice`

model_with_tools = model.bind_tools(tools, tool_choice="auto")

值	含义
`"auto"`	模型自行决定是否调用工具
`"none"`	禁止工具调用
`"required"`	必须调用某个工具
特定工具名或 Provider 格式	强制调用指定工具

具体可用值受 Provider 支持情况影响。

Agent

Agent 是由模型、工具和循环运行时组成的可执行图。模型根据消息决定回答或调用工具；工具结果追加到消息后再次交给模型，直到模型生成最终回答。

创建和调用

from langchain.agents import create_agent
from langchain.messages import HumanMessage

agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt="你是天气助手。",
)

result = agent.invoke({
    "messages": [HumanMessage(content="北京天气如何？")]
})

create_agent() 返回编译后的 LangGraph 图对象，因此支持 invoke、stream、持久化配置和中间件。

agent.invoke() 返回 Agent State 字典，最重要的字段是 messages：

final_message = result["messages"][-1]
answer = final_message.content
message_id = final_message.id
last_call_tokens = final_message.usage_metadata

一次工具调用后的消息序列通常为：

HumanMessage
AIMessage(tool_calls=[...])
ToolMessage(tool_call_id=...)
AIMessage(content="最终回答")

最后一个 AIMessage.usage_metadata 只统计最后一次模型调用。整个 Agent 的 Token 用量需要汇总全部 AIMessage：

from langchain.messages import AIMessage

usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
for message in result["messages"]:
    if isinstance(message, AIMessage) and message.usage_metadata:
        for key in usage:
            usage[key] += message.usage_metadata.get(key, 0)

`create_agent()` 常用参数

参数	意义
`model`	模型字符串或聊天模型对象
`tools`	Tool 对象或可转换为工具的函数列表
`system_prompt`	固定系统提示词
`middleware`	中间件实例列表，按顺序参与 Agent 生命周期
`response_format`	Agent 最终结构化输出策略
`checkpointer`	短期记忆和线程检查点
`store`	跨线程长期记忆存储
`context_schema`	本次调用上下文的类型
`state_schema`	自定义 Agent State 类型

结构化返回

from langchain.agents.structured_output import ToolStrategy

agent = create_agent(
    model=model,
    tools=tools,
    response_format=ToolStrategy(Person),
)

result = agent.invoke({"messages": [{"role": "user", "content": "提取张三的信息"}]})
person = result["structured_response"]

Agent 返回值仍是 State 字典；structured_response 才是 Person 实例。可用策略包括 Provider 原生结构化输出和 ToolStrategy，实际选择取决于模型能力。

流式调用

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "北京天气如何？"}]},
    stream_mode="updates",
):
    print(chunk)

常用模式：

模式	返回内容
`values`	每一步后的完整 Agent State
`updates`	各节点的局部状态更新
`messages`	`(AIMessageChunk, metadata)`
`custom`	工具或节点主动写出的数据
`tasks`、`checkpoints`、`debug`	任务、检查点和调试事件

当 stream_mode 传列表时，每项通常为 (mode, data)。

短期记忆

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(model=model, tools=tools, checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "conversation-1"}}

first = agent.invoke({"messages": [{"role": "user", "content": "我叫小王"}]}, config=config)
second = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, config=config)

同一会话使用相同 thread_id。生产环境使用数据库 Checkpointer。

中间件

中间件在 Agent 生命周期的固定位置读取或修改 State、模型请求、模型响应和工具调用。create_agent(..., middleware=[...]) 按列表顺序安装中间件。

两类 Hook

Node-style

Node-style Hook 在某个阶段运行一次，形式接近 LangGraph 节点：

from langchain.agents.middleware import before_model

@before_model
def inspect_state(state, runtime):
    print(len(state["messages"]))
    return None

返回值：

返回值	意义
`None`	不修改 State，继续执行
`dict`	State 局部更新，由运行时合并
`Command`	更新 State，并可控制跳转

常见阶段包括 before_agent、before_model、after_model 和 after_agent。

Wrap-style

Wrap-style Hook 接收请求与 handler，可在底层调用前后插入逻辑：

from langchain.agents.middleware import wrap_model_call

@wrap_model_call
def log_model_call(request, handler):
    print("before model")
    response = handler(request)
    print("after model")
    return response

必须返回 handler(request) 对应的响应类型。模型包装 Hook 返回模型调用响应，工具包装 Hook 返回 ToolMessage 或运行时接受的工具响应。可以不调用 handler 直接短路，也可以调用多次实现重试，但应明确副作用。

自定义类

from langchain.agents.middleware import AgentMiddleware

class AuditMiddleware(AgentMiddleware):
    def before_model(self, state, runtime):
        print("model input", state["messages"])
        return None

agent = create_agent(model=model, tools=tools, middleware=[AuditMiddleware()])

类适合共享配置或同时实现多个 Hook；单个简单 Hook 使用装饰器更直接。

常用内置中间件

中间件	作用	对运行结果的影响
`SummarizationMiddleware`	历史过长时摘要	替换或压缩传给模型的消息上下文
`HumanInTheLoopMiddleware`	工具执行前审批	Agent 中断，恢复后继续或拒绝工具
`PIIMiddleware`	检测、遮盖或阻止敏感信息	修改消息内容或阻断调用
`TodoListMiddleware`	为复杂任务维护待办项	注入待办工具和相关 State
`ModelCallLimitMiddleware`	限制模型调用次数	达到阈值时终止或报错
`ToolCallLimitMiddleware`	限制工具调用次数	阻止超限调用
`ModelFallbackMiddleware`	主模型失败时切换模型	返回备用模型响应
`LLMToolSelectorMiddleware`	工具很多时先筛选	缩小本轮暴露给模型的工具集合
`ToolRetryMiddleware`	工具失败重试	成功后返回 ToolMessage，耗尽后按策略处理
`ModelRetryMiddleware`	模型失败重试	成功后返回模型响应
`ContextEditingMiddleware`	控制模型上下文	清理或替换发送给模型的消息

中间件修改的是 Agent 执行过程，agent.invoke() 的最外层返回值仍是 Agent State 字典。

人工审批返回关系

HumanInTheLoopMiddleware 需要 Checkpointer。首次调用在受控工具前中断；状态的中断数据会列出工具名称、参数和允许的决策。客户端使用同一 thread_id，通过 Command(resume=...) 提交批准、修改或拒绝结果。

恢复调用返回继续执行后的 Agent State。工具若获批准，会产生 ToolMessage；拒绝或编辑的具体消息内容取决于审批决策和中间件配置。

执行顺序

多个中间件的前置 Hook 通常按注册顺序进入，Wrap-style 的后置部分按嵌套调用的反向顺序退出。中间件有顺序依赖时，应把安全校验、上下文处理和重试顺序写入测试，不要依赖偶然行为。

上下文与记忆

LangChain Agent 中的“上下文”分三类：State 保存本次线程的动态状态，Context 保存单次调用依赖，Store 保存跨线程长期数据。

数据	生命周期	典型内容
Agent State	同一 `thread_id`	消息、当前任务状态、短期记忆
Runtime Context	单次调用	用户 ID、权限、租户、数据库连接标识
Store	跨线程	用户偏好、长期资料、语义记忆

短期记忆

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "thread-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小王"}]},
    config=config,
)

Checkpointer 保存 Agent State。下一次使用相同 thread_id 时，新消息会与已有线程状态配合运行。invoke() 返回最新 State，不返回 Checkpointer 对象。

InMemorySaver 只用于开发；数据库 Checkpointer 用于跨进程恢复。

消息治理

长对话不能无限追加到模型上下文。常见策略：

裁剪：trim_messages() 返回符合 Token 或消息数量限制的新消息列表。
删除：在状态更新中返回 RemoveMessage(id=...)。
摘要：使用 SummarizationMiddleware 将旧消息压缩为摘要。
自定义过滤：在 before_model 中生成本轮模型应看到的消息。

裁剪和摘要影响模型输入，不应误认为删除了 Checkpointer 中的全部历史；是否写回 State 取决于使用方式。

Runtime Context

from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: str
    role: str

agent = create_agent(
    model=model,
    tools=tools,
    context_schema=UserContext,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "查询我的订单"}]},
    context=UserContext(user_id="u-1", role="member"),
)

Context 不会自动加入提示词或 State。中间件与工具需要从 runtime.context 读取它。

长期记忆 Store

Store 使用 namespace 和 key 定位数据：

namespace = ("users", "u-1", "preferences")

store.put(namespace, "language", {"value": "zh-CN"})
item = store.get(namespace, "language")
items = store.search(namespace, query="用户偏好")
store.delete(namespace, "language")

返回值：

方法	返回值
`put(namespace, key, value)`	写入结果通常不作为业务数据使用
`get(namespace, key)`	`Item` 或 `None`
`search(namespace, ...)`	`list[Item]`
`delete(namespace, key)`	删除操作结果通常为空

Item 常用字段包括 namespace、key、value、created_at 和 updated_at。业务数据位于 item.value。

if item is not None:
    language = item.value["value"]

在工具中访问

from langchain.tools import ToolRuntime, tool

@tool
def get_preference(runtime: ToolRuntime[UserContext]) -> str:
    user_id = runtime.context.user_id
    item = runtime.store.get(("users", user_id), "preference")
    return "未设置" if item is None else str(item.value)

runtime 是注入参数，不应让模型生成。工具对模型可见的 Schema 中只保留业务参数。

LangSmith

LangSmith 用于记录、查看和评估 LangChain/LangGraph 运行。它不替代模型或 Agent，也不会改变 invoke() 的业务返回类型。

开启追踪

通常设置以下环境变量：

LANGSMITH_TRACING=true
LANGSMITH_API_KEY=<YOUR_API_KEY>
LANGSMITH_PROJECT=my-project

不同版本可能使用兼容的旧变量名。API Key 不应写入代码或提交到版本库。

标记运行

response = agent.invoke(
    {"messages": [{"role": "user", "content": "你好"}]},
    config={
        "run_name": "customer-service",
        "tags": ["production", "customer-service"],
        "metadata": {"user_id": "u-1"},
    },
)

run_name、tags 和 metadata 会进入 Trace，便于筛选。返回值仍是 Agent State；LangSmith Trace ID 不会自动替换业务返回结果。

能看到什么

每次模型、工具、Retriever 和 Agent 调用的输入输出
调用层级、耗时、异常和重试
模型 Token 用量与成本，前提是 Provider 返回相关元数据
工具名称、参数与 ToolMessage
标签、项目和自定义元数据

生产环境写入 metadata 时避免放入密码、密钥、完整身份证号等敏感信息。需要记录用户标识时优先使用内部 ID 或脱敏值。

Comments

Notes, questions, and follow-ups are welcome here.

Comments are temporarily unavailable because WALINE_SERVER_URL or PUBLIC_WALINE_SERVER_URL is not configured.

模型创建与调用

创建模型

invoke()

stream()

batch() 与 batch_as_completed()

异步方法

config 参数

消息与提示词

消息类型

ChatPromptTemplate

消息占位符

partial()

结构化输出

Pydantic

TypedDict、JSON Schema 与 dataclass

保留原始消息

与 Agent 的区别

工具

定义工具

自定义参数 Schema

绑定模型

手动执行完整流程

tool_choice

Agent

创建和调用

create_agent() 常用参数

结构化返回

流式调用

短期记忆

中间件

两类 Hook

Node-style

Wrap-style

自定义类

常用内置中间件

人工审批返回关系

执行顺序

上下文与记忆

短期记忆

消息治理

Runtime Context

长期记忆 Store

在工具中访问

LangSmith

开启追踪

标记运行

能看到什么

`invoke()`

`stream()`

`batch()` 与 `batch_as_completed()`

`config` 参数

`ChatPromptTemplate`

`partial()`

`tool_choice`

`create_agent()` 常用参数