搜索

Resource / Python

LangChain
Cheatsheet

集中整理 LangChain Python Agent 开发中常用的模型调用、Tools、Memory、Structured Output 和 Middleware。示例基于当前官方稳定文档整理。

参考版本
LangChain 1.4.0
语言范围
Python 3.10+
核心 API
create_agent
INPUTmessages
MODELreason + choose
TOOLSact → observe
OUTPUTfinal response
MIDDLEWAREbefore · wrap · after
LangChain Agent 的核心循环：模型决定是否调用工具；Middleware 可以在模型调用、工具执行等阶段加入额外处理逻辑。
一句话理解

LangChain 提供统一的模型、消息、Tool 和 Agent 接口。Middleware 可用于处理 Memory、安全检查、重试和动态上下文，create_agent 底层使用 LangGraph 运行。

01 / SCOPE

范围、安装与迁移

基础包要求 Python 3.10+。模型厂商集成独立发布，只安装实际使用的集成包即可，不要假设 langchain 会捆绑模型 SDK。

terminalBash
pip install -U langchain
pip install -U langchain-openai  # 示例：OpenAI provider
langchain

Agent 核心：模型、消息、工具、middleware 与 create_agent。

langchain-*

模型、向量库与第三方服务的独立集成包。

langgraph

Agent 底层运行时，以及 checkpointer、store、interrupt 等能力。

langchain-classic

仅为仍需旧 Chains、旧 Retrievers 等 API 的迁移项目。

02 / AGENT

最小可运行 Agent

agent.pyPython
from langchain.agents import create_agent
from langchain.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """Return the current weather for a city."""
    return f"{city}: sunny, 24°C"
 
agent = create_agent(
    model="openai:gpt-5.4",
    tools=[get_weather],
    system_prompt="You are a concise weather assistant.",
)
 
result = agent.invoke({
    "messages": [
        {"role": "user", "content": "上海天气怎么样？"}
    ]
})
print(result["messages"][-1].text)

create_agent 返回一个已编译的 LangGraph。模型会在“调用模型 → 执行工具 → 把结果交还模型”的循环中运行，直到模型不再请求工具。普通同步程序使用 invoke；异步 I/O 使用 ainvoke。

03 / MODEL

Models：单次调用、Streaming 与工具绑定

需要统一切换模型厂商时，使用 init_chat_model；需要模型厂商特有参数时，使用对应的 Chat Model 类。超时、重试和采样参数以具体集成支持的范围为准。

model.pyPython
from langchain.chat_models import init_chat_model
 
model = init_chat_model(
    "openai:gpt-5.4",
    temperature=0,
    timeout=30,
    max_retries=3,
)
 
reply = model.invoke("用一句话解释 RAG")
print(reply.text)
 
for chunk in model.stream("给我三个 Agent 测试要点"):
    print(chunk.text, end="", flush=True)
模型调用方式
方法返回适用
invoke()AIMessage一次输入，等待完整结果
stream()AIMessageChunk 迭代器逐步呈现文本、工具调用或推理块
batch()结果列表多个互相独立的输入
ainvoke / astream / abatch对应的异步方法异步服务与高并发 I/O

常用配置包括：用 bind_tools([...]) 绑定工具，用 with_structured_output(Schema) 获取解析后的结构化结果。具体功能是否可用取决于所选模型和模型厂商。

04 / MESSAGE

Messages：统一的模型上下文

messages.pyPython
from langchain.messages import (
    SystemMessage,
    HumanMessage,
    AIMessage,
    ToolMessage,
)
 
messages = [
    SystemMessage("Answer with cited evidence."),
    HumanMessage("What changed?"),
]
response = model.invoke(messages)
 
print(response.text)            # 归一化文本访问
print(response.content_blocks)  # 文本、推理、工具调用等标准块
print(response.usage_metadata)  # 可用时包含 token 用量
System / Developer

应用给模型的行为指令，具体支持哪些角色以模型厂商文档为准。

Human / User

用户输入，可包含文本或多模态内容块。

AI / Assistant

模型输出，可能同时携带文本、工具调用、用量和响应元数据。

Tool

工具执行结果，其中的 tool_call_id 必须与模型发出的 Tool Call 对应。

05 / TOOL

Tools 与 ToolRuntime

简单函数可直接放进 tools，需要自定义名称、schema 或运行时访问时使用 @tool。ToolRuntime 参数由框架注入，不会暴露在模型看到的工具 schema 中。

tools.pyPython
from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime
 
@dataclass
class Context:
    user_id: str
 
@tool
def lookup_order(order_id: str, runtime: ToolRuntime[Context]) -> str:
    """Look up an order the current user is allowed to access."""
    user_id = runtime.context.user_id
    order = load_authorized_order(user_id, order_id)
    return order.model_dump_json()
 
agent = create_agent(
    model="openai:gpt-5.4",
    tools=[lookup_order],
    context_schema=Context,
)
 
agent.invoke(
    {"messages": [{"role": "user", "content": "查询订单 A-42"}]},
    context=Context(user_id="u-7"),
)
模型参数

参数

只放需要由模型决定的业务参数，类型和描述要具体。

运行时注入

Runtime

读取 state、context、store、stream writer、config 与 call ID。

业务校验

校验

工具内部必须重新检查权限、金额、路径，以及写数据库、发送消息等操作。

model_with_tools.pyPython
model_with_tools = model.bind_tools([lookup_order])
response = model_with_tools.invoke("查询订单 A-42")
 
# 直接使用 model 时，调用、执行工具、追加 ToolMessage 的闭环由你负责
for call in response.tool_calls:
    print(call["name"], call["args"], call["id"])
 
# create_agent 会替你运行这段工具循环

06 / SCHEMA

Structured Output：返回结构化数据

structured_agent.pyPython
from pydantic import BaseModel, Field
from langchain.agents import create_agent
 
class Ticket(BaseModel):
    category: str = Field(description="billing, account, or technical")
    priority: int = Field(ge=1, le=5)
    summary: str
 
agent = create_agent(
    model="openai:gpt-5.4",
    tools=[],
    response_format=Ticket,
)
 
result = agent.invoke({
    "messages": [{"role": "user", "content": "付款两次，尽快处理"}]
})
ticket: Ticket = result["structured_response"]

向 create_agent(response_format=...) 直接传 schema 时，如果模型支持原生结构化输出，LangChain 会优先使用；否则改用工具调用策略。最终对象位于 structured_response。

standalone_model.pyPython
structured_model = model.with_structured_output(Ticket)
ticket = structured_model.invoke("付款两次，尽快处理")
Schema 选择
Schema优势适用
Pydantic字段描述、验证、嵌套模型默认首选的 Python 业务数据模型
TypedDict轻量、静态类型友好不需要运行时模型对象
dataclass标准库数据对象已有 dataclass 领域模型
JSON Schema语言无关跨系统共享数据结构

07 / CONTEXT

Context、State 与 Store 不要混用

单次调用

Context

提供用户 ID、权限、数据库连接等本次调用需要的依赖。

当前 Thread

State

保存消息、当前计划和本对话中会变化的运行数据。

跨 Thread

Store

保存跨会话偏好、长期 Memory 与共享业务数据。

runtime_context.pyPython
from dataclasses import dataclass
from langchain.agents import create_agent
 
@dataclass
class Context:
    user_id: str
    db: "Database"
 
agent = create_agent(
    model="openai:gpt-5.4",
    tools=[lookup_order],
    context_schema=Context,
)
 
agent.invoke(
    {"messages": [{"role": "user", "content": "我的订单呢？"}]},
    context=Context(user_id="u-7", db=db),
)

模型每次调用实际使用的 prompt、messages、tools、model 和 response format 只服务于当前请求。State 保存当前 Thread 中会变化的数据，Store 保存需要跨 Thread 使用的数据。

08 / MEMORY

短期 Memory 与长期 Memory

短期：同一个 thread 的状态

short_term_memory.pyPython
from langgraph.checkpoint.memory import InMemorySaver
 
agent = create_agent(
    model="openai:gpt-5.4",
    tools=[],
    checkpointer=InMemorySaver(),  # 本地开发 / 测试
)
 
config = {"configurable": {"thread_id": "conversation-42"}}
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小林"}]},
    config=config,
)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config=config,
)

checkpointer 在每个执行步骤保存 State；后续调用必须继续使用同一个 thread_id。InMemorySaver 适合本地示例和测试，生产环境使用数据库后端。

terminal · PostgreSQLBash
pip install -U langgraph-checkpoint-postgres "psycopg[binary,pool]"

长期：跨 thread 的 Store

long_term_memory.pyPython
from langgraph.store.memory import InMemoryStore
 
store = InMemoryStore()  # 生产环境换成持久化 Store
agent = create_agent(
    model="openai:gpt-5.4",
    tools=[read_preference, save_preference],
    store=store,
)
 
# 工具通过 runtime.store，以 namespace + key 读写跨 thread 数据

09 / MIDDLEWARE

Middleware：统一处理 Agent 运行过程中的公共逻辑

Middleware 可以在 Agent 启动、模型调用和工具执行前后插入处理逻辑。优先组合官方内置 Middleware，现有能力无法满足业务规则时再编写自定义 hook。

middleware.pyPython
from langchain.agents import create_agent
from langchain.agents.middleware import (
    SummarizationMiddleware,
    ToolRetryMiddleware,
    ModelFallbackMiddleware,
)
 
agent = create_agent(
    model="openai:gpt-5.4",
    tools=tools,
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-5.4-mini",
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
        ToolRetryMiddleware(max_retries=3),
        ModelFallbackMiddleware("anthropic:claude-sonnet-4-6"),
    ],
)
常见控制需求
需求优先能力目的
上下文过长SummarizationMiddleware压缩旧消息，保留近期上下文
瞬时故障Tool / Model retry对可重试错误应用退避
模型不可用ModelFallbackMiddleware按顺序尝试备用模型
成本失控Model / Tool call limit限制单次运行调用次数
敏感操作HumanInTheLoopMiddleware执行前暂停并等待决策
敏感数据PIIMiddlewareblock、redact、mask 或 hash

动态 prompt

dynamic_prompt.pyPython
from langchain.agents.middleware import dynamic_prompt, ModelRequest
 
@dynamic_prompt
def user_prompt(request: ModelRequest) -> str:
    tier = request.runtime.context.account_tier
    return f"Help the user. Account tier: {tier}."
 
agent = create_agent(
    model=model,
    tools=tools,
    middleware=[user_prompt],
    context_schema=Context,
)

10 / STREAM

Streaming v2：统一事件形状

stream.pyPython
for part in agent.stream(
    {"messages": [{"role": "user", "content": "分析这份订单"}]},
    stream_mode=["updates", "messages", "custom"],
    version="v2",
):
    if part["type"] == "updates":
        print("step:", part["data"])
    elif part["type"] == "messages":
        token, metadata = part["data"]
        print(token.text, end="")
    elif part["type"] == "custom":
        print("progress:", part["data"])
updates

每个 Agent 步骤完成后的 State 更新，适合展示“模型 / 工具正在做什么”。

messages

LLM token 与 metadata，适合逐字渲染回答、推理或工具调用块。

custom

业务自定义进度，例如导入 20 / 100 条记录。

version="v2"

统一返回含 type、ns、data 的 StreamPart。

tool_progress.pyPython
@tool
def import_orders(runtime: ToolRuntime) -> str:
    """Import pending orders."""
    runtime.stream_writer({"done": 20, "total": 100})
    # ...
    return "Imported 100 orders"

11 / RETRIEVE

Retrieval / RAG：从文档切分到检索

rag_tool.pyPython
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain.tools import tool
 
chunks = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    add_start_index=True,
).split_documents(documents)
 
vectorstore = InMemoryVectorStore.from_documents(
    chunks,
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
 
@tool
def retrieve(query: str) -> str:
    """Search the approved knowledge base for relevant passages."""
    docs = retriever.invoke(query)
    return "
 
".join(
        f"SOURCE: {doc.metadata}
{doc.page_content}" for doc in docs
    )
LOADSPLITEMBEDSTORERETRIEVE

将 retriever 封装成工具，适合由 Agent 自主决定是否检索的 RAG（Agentic RAG）。如果应用始终需要检索，可以直接调用 retriever，再把文档交给模型。k、chunk 大小和 overlap 没有通用最优值，应根据真实语料和评估集调整。

  • 保留 source、页码、URL、更新时间与访问级别等 metadata。
  • 检索前执行身份与权限过滤，不要只在 prompt 中要求模型忽略无权内容。
  • 把引用与答案分开验证；“检索到了”不等于“答案受证据支持”。
  • 文档加载器、embedding 和向量库通常位于独立集成包，按实际后端安装。

12 / SAFETY

Guardrails 与 Human-in-the-loop

固定规则适合检查权限、格式、金额上限和敏感字段；基于模型判断的 Guardrail 适合语义检查，但会增加延迟、成本和结果波动。退款、写数据库、发送消息等高风险操作，应在工具执行前做强制校验或人工审批。

human_review.pyPython
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
 
agent = create_agent(
    model="openai:gpt-5.4",
    tools=[read_order, issue_refund],
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={"read_order": False, "issue_refund": True}
    )],
    checkpointer=InMemorySaver(),
)
 
config = {"configurable": {"thread_id": "review-42"}}
pending = agent.invoke(request, config=config)
 
completed = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
)

13 / VERIFY

Tracing、测试与评估

Agent 输出虽然存在随机性，但仍然可以通过分层测试保证关键行为。测试可以分成单元测试、真实集成测试和执行过程评估，分别检查业务规则、外部集成和 Agent 的整体表现。

test_agent.pyPython
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain.messages import AIMessage
 
fake = GenericFakeChatModel(messages=iter([
    AIMessage(content="expected answer"),
]))
agent = create_agent(fake, tools=[])
 
result = agent.invoke({"messages": [{"role": "user", "content": "test"}]})
assert result["messages"][-1].text == "expected answer"
Unit

用 GenericFakeChatModel、假工具与 InMemorySaver 验证路由、状态和异常分支。

Integration

调用真实模型和外部服务，检查返回结构、schema 和必须满足的业务条件，不要逐字匹配自然语言。

Trajectory eval

判断工具选择、调用顺序和最终任务完成度；关键数据集进入回归测试。

Trace

记录模型输入、工具参数、延迟、token、错误与重试；生产环境先定义敏感数据策略。

terminal · tracingBash
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=...
export LANGSMITH_PROJECT=my-agent

14 / IMPORTS

当前导入路径速查

核心 Python API
能力导入路径
Agentfrom langchain.agents import create_agent, AgentState
Modelsfrom langchain.chat_models import init_chat_model
Messagesfrom langchain.messages import HumanMessage, AIMessage, ToolMessage
Toolsfrom langchain.tools import tool, ToolRuntime
Middlewarefrom langchain.agents.middleware import ...
Embeddingsfrom langchain.embeddings import init_embeddings
Text splitterfrom langchain_text_splitters import RecursiveCharacterTextSplitter
Provider 类from langchain_openai import ChatOpenAI, OpenAIEmbeddings
Checkpointfrom langgraph.checkpoint.memory import InMemorySaver
Storefrom langgraph.store.memory import InMemoryStore
Resumefrom langgraph.types import Command
Legacy chainsfrom langchain_classic.chains import ...

15 / DEBUG

常见错误索引

模型不调用工具

检查工具名、docstring、参数 schema、模型是否支持 tool calling，以及 prompt 是否给出了清晰触发条件。

消息被模型服务拒绝

确认每个 AI Tool Call 后都有相同 ID 的 ToolMessage，且消息顺序完整。

下一轮忘记上下文

确认 agent 配置了 checkpointer，并在每次调用中复用同一 thread_id。

结构化输出失败

缩小 schema，补充字段描述，确认模型能力；将验证错误当成可观测事件处理。

流式没有 token

确认使用 messages 模式、调用链支持 streaming，并正确解析所选 streaming 版本。

上下文持续膨胀

设置 summarization / trimming 策略，区分短期 State 与真正需要长期保存的 Store 数据。

工具重试导致重复写入

写操作使用幂等键；只重试明确可重试的异常，并记录重试次数与外部请求 ID。

升级后导入失败

先查 v1 migration guide；新项目不要用 langchain-classic 继续沿用旧 API。

16 / SHIP

生产检查清单

  • 锁定并定期升级 langchain、模型厂商集成与持久化相关包；升级前运行回归测试。
  • 对每次模型与外部 I/O 设置 timeout；重试只覆盖瞬时失败并使用退避。
  • 所有写工具具备权限校验、参数验证、幂等键和可审计的外部请求 ID。
  • 高风险工具启用人工审批；恢复操作复用原 thread_id 并验证决策类型。
  • 生产 checkpointer / store 使用持久化后端，验证迁移、加密、备份、恢复和数据删除。
  • 定义消息裁剪或摘要阈值，同时监控 token、延迟、工具失败率、重试和成本。
  • 日志与 trace 在写入前处理 PII、密钥和工具返回；控制访问与保留周期。
  • 用 fake model 覆盖固定分支，用真实模型厂商测试 schema 与工具集成，用 eval 检查执行过程。
  • RAG 记录 source metadata，执行权限过滤，并用真实问题集评估检索与答案引用。
  • 为 rate limit、模型厂商故障、空检索、无效工具参数、中断恢复和 Streaming 断线准备备用模型、重试或错误提示。

17 / SOURCES

官方资料与参考范围

版本号来自 PyPI；安装、导入路径、API 用法与推荐实践来自 LangChain 当前官方文档。页面中的模型名只用于演示初始化方式，实际可用的模型和参数以所选模型厂商的文档为准。