【LangChain】二.LangChain v1.0-高级特性（记忆、RAG 、Multi-Agent ）

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver


agent = create_agent(
    model="openai:gpt-5.4",
    tools=[],
    checkpointer=InMemorySaver(),
)

短期记忆不是额外外挂，而是 agent 状态如何跨轮次被保存。

但短期记忆不等于"把所有历史消息无限堆进去"。长线程必须做三件事：

裁剪
摘要
只保留真正影响后续决策的信息

这也是为什么短期记忆会和 SummarizationMiddleware、ContextEditingMiddleware 强绑定。

1.2 长期记忆：跨线程持久化

长期记忆对应的是 store。它不是当前线程的一部分，而是跨线程、跨会话可复用的持久存储。

官方文档常提三类长期记忆：

semantic memory：用户资料、偏好、事实性知识
episodic memory：过去发生过的任务、事件、交互片段
procedural memory：做事规则、操作偏好、固定流程

对开发者来说，更关键的不是记住术语，而是看明白：长期记忆解决的是"下次还能记得"，不是"本轮不要忘"。

from dataclasses import dataclass
from langchain.tools import ToolRuntime, tool


@dataclass
class Context:
    user_id: str


@tool
def get_user_preference(runtime: ToolRuntime[Context]) -> str:
    item = runtime.store.get(("preferences",), runtime.context.user_id)
    return str(item.value) if item else "No saved preference"

如果要写入长期记忆，也是同样的入口：

@tool
def save_user_preference(key: str, value: str, runtime: ToolRuntime[Context]) -> str:
    runtime.store.put(("preferences",), runtime.context.user_id, {key: value})
    return "Preference saved"

长期记忆在 LangChain 里不是一个聊天外挂，而是一个标准化的可读写存储接口。

1.3 记忆系统真正难的地方

记忆系统最难的地方，从来不是"怎么存"，而是三件事：

写什么
什么时候写
什么时候读

如果什么都写，store 会变垃圾场。如果什么都读，模型上下文会被历史噪声淹没。所以记忆系统的设计，本质上不是存储问题，而是筛选问题。

2. RAG / 知识接入

2.1 RAG 和 memory 的边界

这两个概念经常被混写，但你必须分清。

memory 更偏"系统是否记得用户、记得过去交互、记得偏好和经验"
retrieval 更偏"系统是否能按需拿到外部知识"

一个是"过去沉淀下来的内部记忆"，一个是"当前按需拉取的外部知识"。这就是它们应该分章讲的原因。

RAG 要解决的问题是：模型知识是静态的，上下文窗口是有限的，外部知识怎么在需要的时候进来？

详细见【LangChain】LangChain v1.0 实战：从零构建 RAG 应用。

3. Multi-agent

Multi-agent 回答的问题是：单个 agent 装不下时，怎么拆？

这章还是先给结论：

多代理很重要，但它不是默认起点。

很多时候，一个带动态工具、动态 prompt、合理 middleware 的单 agent，已经足够解决问题。
只有在这些信号很明显时，多代理才真正有必要：

单 agent 的上下文已经严重过载，提示词和工具描述越堆越大
不同任务需要完全不同的角色、知识域和工具集
不同团队要独立维护不同能力模块
你明确需要并行执行、阶段切换，或者强约束流程

3.1 多代理到底在解决什么

LangChain 官方文档对这件事的判断非常准确：开发者口中的 multi-agent，通常想解决的是上下文管理、分工维护和并行执行，而不只是"让系统更聪明"。

更直白一点说：

如果上下文无限大、延迟无限低，你完全可以把所有知识和工具都塞进一个 agent
但现实不是这样，所以你必须决定：谁看到什么，谁在什么时候出手，谁能调用哪些工具

这就是官方反复强调的 context engineering。
多代理设计的核心，不是"多几个角色名"，而是把正确的信息交给正确的执行单元。

3.2 先把五种模式分清，再谈实现

旧写法里最容易让读者混淆的一点，是只记住了几个术语，但不知道它们的边界。
LangChain 官方 Multi-agent 一共讲了五种模式：

模式	核心机制	最适合什么问题	主要代价
`subagents`	主代理把子代理当工具来调用	中央调度、多领域隔离、强控制	多一次协调调用，链路更长
`handoffs`	工具更新状态，切换当前活跃角色或阶段	多轮对话、顺序约束、当前角色直接和用户对话	状态设计更关键，消息传递更敏感
`skills`	单 agent 按需加载专门技能提示和资料	大量专业能力、按需加载上下文	不像子代理那样天然隔离执行过程
`router`	先路由分类，再把任务派给一个或多个专门代理	输入类别清晰、需要并行查询多个领域	路由本身通常是额外一步
`custom workflow`	用 LangGraph 自定义图，混合确定性节点和 agent 节点	标准模式不够用，需要分支、循环、人工审批、复杂编排	代码更多，但控制力最强

这个表最好反复看。
因为后面所有"怎么写代码"，其实都是在回答：你到底是要中央调度、状态切换、按需加载、先分类，还是完全自定义流程？

3.3 `subagents`：主代理当 supervisor，子代理当工具

subagents 是最符合很多人直觉的多代理模式：
有一个主代理负责和用户对话、保留主线程记忆、决定调用谁；子代理只处理自己那一小块任务。

官方文档里对它的定义有几个关键词，非常重要：

centralized control：所有路由都经过主代理
subagents via tools：子代理通常是以 tool 的形式暴露给主代理
stateless subagents：子代理默认是无状态的，记忆主要由主代理持有
parallel execution：主代理可以在一个回合里并行调用多个子代理

什么时候优先考虑 subagents？

你有多个明确领域，比如日历、邮件、CRM、数据库、代码库
子代理不需要直接和用户对话
你希望主代理统一掌控流程和结果整合
你非常在意上下文隔离，不想把每个领域的长上下文都塞进主对话

一个最小实现通常就是：先创建子代理，再把子代理包装成工具，最后让主代理决定什么时候调用它们。

from langchain.agents import create_agent
from langchain.tools import tool


research_agent = create_agent(
    model="openai:gpt-5.4",
    tools=[],
    system_prompt="You are a research specialist. Focus on facts, sources, and background.",
)

writer_agent = create_agent(
    model="openai:gpt-5.4",
    tools=[],
    system_prompt="You are a writing specialist. Turn notes into clean drafts.",
)


@tool("research", description="Research facts, sources, and background information")
def call_research_agent(query: str) -> str:
    result = research_agent.invoke(
        {"messages": [{"role": "user", "content": query}]}
    )
    return result["messages"][-1].content


@tool("write_draft", description="Write or revise article drafts")
def call_writer_agent(task: str) -> str:
    result = writer_agent.invoke(
        {"messages": [{"role": "user", "content": task}]}
    )
    return result["messages"][-1].content


supervisor = create_agent(
    model="openai:gpt-5.4",
    tools=[call_research_agent, call_writer_agent],
    system_prompt=(
        "You are the supervisor. "
        "Use research for fact-finding and write_draft for drafting. "
        "You keep the conversation with the user and combine the results."
    ),
)

这段代码背后的开发含义比代码本身重要：

主代理负责"调度"和"整合"
子代理负责"在隔离上下文里完成局部任务"
主线程不需要吃下所有中间推理细节，只需要拿回结果摘要

官方文档还特别强调了几个设计点，实际开发里非常有用：

第一，工具暴露方式有两种。

tool per agent：一个子代理一个工具，最直观，也最容易细调输入输出
single dispatch tool：用一个统一的 task(agent_name, description) 调度多个子代理，适合代理很多、团队很多、注册表可能动态变化的场景

如果代理很多，官方建议不要把所有代理都硬编码进主 prompt。
你可以再提供一个 list_agents 或 search_agents 工具，让主代理按需发现可用子代理，这本质上也是 progressive disclosure。

第二，同步还是异步，要按业务链路决定。

同步调用：主代理必须等子代理结果回来，最简单
异步调用：主代理只发起后台任务，不阻塞当前对话

官方给异步模式的建议非常实用：通常做成三类工具即可。

start_job
check_status
get_result

这说明一件事：这里的 async 不是 Python 语法层面的 async/await，而是产品层面的后台任务设计。

第三，subagents 的关键不是"多几个 agent"，而是上下文工程。

你至少要认真设计三件事：

子代理的名字和描述：这是主代理判断何时调用它的主要提示信号
传给子代理的输入：只传 query，还是连任务约束、用户偏好、摘要一起传
子代理返回什么：是返回最终摘要，还是返回完整对话历史

绝大多数时候，返回面向主代理的简洁结果比返回完整 transcript 更好。
因为你做 subagents 的初衷，本来就是为了隔离上下文，而不是把所有上下文重新灌回主线程。

3.4 `handoffs`：让"当前活跃角色"发生切换

handoffs 是另一种很常见、也很容易被误解的模式。
它的核心不是"神秘代理协议"，而是：

工具调用更新状态，状态变化再触发 prompt、工具集或执行节点的变化。

官方文档里讲得很清楚，handoffs 的关键词是：

state-driven behavior：行为由状态决定
tool-based transitions：状态切换通常由工具触发
direct user interaction：切换后的角色可以继续直接和用户对话
persistent state：这个状态会跨回合保留

所以它特别适合什么？

你要做多阶段对话
你要强制顺序约束，先收集信息，再开放后续能力
当前阶段的角色需要直接和用户继续聊下去

比如客服流里，必须先拿到保修编号，再进入退款或维修流程。
这种场景用 handoffs 就比 subagents 更自然。

下面是一个简化版思路：工具写状态，middleware 根据状态切换当前阶段的 prompt 和工具集。

from typing import Callable
from typing_extensions import TypedDict
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRequest, ModelResponse, wrap_model_call
from langchain.messages import ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


class SupportState(TypedDict, total=False):
    current_step: str
    warranty_status: str


@tool
def record_warranty_status(
    status: str,
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"Warranty status recorded: {status}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "warranty_status": status,
            "current_step": "specialist",
        }
    )


@wrap_model_call
def apply_step_config(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    step = request.state.get("current_step", "triage")
    configs = {
        "triage": {
            "prompt": "Collect warranty information before solving the issue.",
            "tools": [record_warranty_status],
        },
        "specialist": {
            "prompt": "Provide a solution based on warranty_status={warranty_status}.",
            "tools": [],
        },
    }
    config = configs[step]
    request = request.override(
        system_prompt=config["prompt"].format(**request.state),
        tools=config["tools"],
    )
    return handler(request)


agent = create_agent(
    model="openai:gpt-5.4",
    tools=[record_warranty_status],
    state_schema=SupportState,
    middleware=[apply_step_config],
    checkpointer=InMemorySaver(),
)

这段代码里有两个官方文档特别提醒的细节，很多人第一次写会漏：

handoff 工具通常返回 Command，因为它要更新状态，甚至跳转节点
如果工具更新了消息历史，必须补上和当前 tool_call_id 对应的 ToolMessage，否则消息序列会不完整

这个模式还有两种落地方式：

单 agent + middleware：本质上是一个 agent，根据状态动态换 prompt 和 tools
多个 agent subgraphs：每个角色是一个单独节点，handoff 工具通过 Command(goto=...) 跳到另一个节点

前者实现更轻，后者边界更清楚。
但后者也更考验你对消息传递和上下文裁剪的掌控，因为不同子图之间传什么消息，得你自己明确决定。

3.5 `skills`：不是换一个 agent，而是让同一个 agent 按需加载专长

skills 是很多人容易忽视、但实际上非常实用的一种模式。
它的想法不是"多造几个代理"，而是：

让一个主 agent 在需要时加载某项专业能力对应的提示、知识、模板或参考资料。

官方把它概括成几个关键词：

prompt-driven specialization
progressive disclosure
lightweight composition
reference awareness

这意味着 skills 更像什么？
更像是"按需加载的专长包"，而不是一个完整的、独立执行的子代理。

什么时候优先用 skills？

你还是想维持一个主 agent
专长很多，但不是每次都要全部塞进上下文
这些能力之间没有很强的流程切换约束
你想让不同团队独立维护不同技能文件或知识包

一个最小实现通常是：做一个 load_skill 工具，按名字返回该技能的提示和上下文。

from pathlib import Path
from langchain.agents import create_agent
from langchain.tools import tool


SKILLS = {
    "write_sql": Path("skills/write_sql.md"),
    "review_legal_doc": Path("skills/review_legal_doc.md"),
}


@tool
def load_skill(skill_name: str) -> str:
    """Load a specialized skill prompt and references."""
    return SKILLS[skill_name].read_text(encoding="utf-8")


agent = create_agent(
    model="openai:gpt-5.4",
    tools=[load_skill],
    system_prompt=(
        "You are a helpful assistant. "
        "When the task becomes specialized, use load_skill to load the right skill first."
    ),
)

这段代码真正表达的是：把专业知识做成按需读取的资产，而不是默认塞进 system prompt。

官方文档还给了几个很好的扩展方向：

动态工具注册：加载某个 skill 时，不只是注入提示，也动态开放这类工具
分层技能：先加载 data_science，再按需加载 pandas_expert 或 visualization
引用外部资产：skill 本身很短，但它告诉 agent 相关脚本、模板、文档放在哪里，需要时再读

所以 skills 的核心收益是：少塞上下文，按需揭示能力。
它本质上仍然是 context engineering。

3.6 `router`：先做分类，再并行派发

router 模式特别适合"输入类别很清楚"的系统。
它不是让一个主代理边聊边决定，而是先做一次路由判断，再把任务发给一个或多个专门代理。

官方对它的定义很清楚：

先做 routing / classification
调用零个、一个或多个专门代理
最后把结果综合成一个答案

它最适合什么？

领域边界清楚，比如 GitHub、Notion、Slack、数据库、代码仓库
你需要并行查询多个来源
你希望路由逻辑更显式、更容易替换成规则或轻量模型

LangGraph 里，单路由和并行 fan-out 的写法分别对应 Command 和 Send。
并行版思路很简单：

from typing import TypedDict
from langgraph.types import Send


class RouterState(TypedDict):
    query: str


def classify_query(query: str) -> list[dict[str, str]]:
    """Return a list like [{'agent': 'github_agent', 'query': '...'}]."""
    ...


def route_query(state: RouterState):
    classifications = classify_query(state["query"])
    return [
        Send(item["agent"], {"query": item["query"]})
        for item in classifications
    ]

这个模式和 subagents 很像，但不要混掉：

router 更像一次显式的前置分类步骤
subagents 则是主代理在持续对话里动态决定调用谁

官方还专门提醒了 router 的两种形态：

stateless router：每次请求都重新路由，不保留会话状态
stateful router：需要显式持久化上下文

真实项目里，一个很好用的折中方案是：把无状态 router 封装成一个工具，再交给会话型 agent 去调用。
这样，记忆由外层 agent 负责，router 本身保持简单。

3.7 `custom workflow`：标准模式不够时，直接上 LangGraph

如果你看到这里已经觉得：
"我这个系统既有检索、又有审批、还要并行、还要失败重试、还要分支和循环，好像哪种模式单独都不够。"
那就对了，这正是 custom workflow 的适用范围。

官方对它的定义非常直接：
用 LangGraph 自己定义执行图，把确定性逻辑和 agent 行为混在同一个工作流里。

它适合什么？

标准模式已经装不下真实业务
你需要分支、循环、并行、人工中断、审批、回滚
某些步骤必须是确定性代码，某些步骤才交给模型

最关键的一点是：LangChain agent 可以直接作为 LangGraph 的一个节点来调用。

from typing import TypedDict
from langchain.agents import create_agent
from langgraph.graph import StateGraph, START, END


class WorkflowState(TypedDict):
    question: str
    documents: list[str]
    answer: str


agent = create_agent(
    model="openai:gpt-5.4",
    tools=[],
)


def retrieve(state: WorkflowState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"documents": docs}


def answer(state: WorkflowState) -> dict:
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": f"问题：{state['question']}\n\n参考资料：{state['documents']}",
                }
            ]
        }
    )
    return {"answer": result["messages"][-1].content}


workflow = (
    StateGraph(WorkflowState)
    .add_node("retrieve", retrieve)
    .add_node("answer", answer)
    .add_edge(START, "retrieve")
    .add_edge("retrieve", "answer")
    .add_edge("answer", END)
    .compile()
)

这个例子已经说明了 custom workflow 的价值：

retrieve 是确定性节点
answer 是 agent 节点
状态通过图中的结构化字段传递

从官方文档的角度看，router 本身就是一种 custom workflow。
真正的分界线不是"有没有多个 agent"，而是你是不是需要自己掌控整张执行图。

3.8 怎么选模式

官方文档还做了性能对比，这里不需要死记数字，但要记住结论：

单次 one-shot 请求：handoffs、skills、router 往往比 subagents 少一次协调开销
重复请求：handoffs 和 skills 更容易复用当前状态或已加载技能
多领域并行任务：subagents 和 router 更有优势，因为它们更适合并行和上下文隔离

把它翻译成开发决策，就是：

你要中央调度 + 强隔离，选 subagents
你要当前角色直接和用户继续聊 + 阶段切换，选 handoffs
你要一个主 agent + 按需加载专长，选 skills
你要显式分类 + 并行派发多个领域，选 router
你要完全控制分支、循环、人工环节和确定性步骤，选 custom workflow

3.9 什么时候该用 Multi-agent

多代理不是默认选项，而是解决特定问题的手段。

适合用 Multi-agent 的场景：

任务类型差异大，需要不同专业角色
单个 Agent 上下文过载，需要隔离
不同团队要独立维护不同能力
需要明确的职责边界和审计追踪

不适合用 Multi-agent 的场景：

任务相对简单，单 Agent 能搞定
上下文长度可控，不需要拆分
团队规模小，维护成本考虑
刚开始探索，先验证核心逻辑

建议： 先从单 Agent 开始，遇到问题再考虑拆分。

4. 总结：LangChain 全景地图

现在把上下篇的内容收成一张完整的开发地图。

4.1 核心能力回顾

模块	职责	关键对象
核心组件	让 Agent 跑起来：模型、消息、装配入口、流式观察、结构化输出	model、messages、agent、streaming、structured output
工具系统	让 Agent 能行动，并把很多高级能力重新拉回同一条主轴	Tool、ToolRuntime、Command、MCP
中间件与控制系统	让 Agent 可控、可治理、可上线	middleware、runtime、context engineering
记忆系统	让 Agent 在当前线程和跨线程都能保留有效状态	checkpointer、store
RAG / 知识接入	让外部知识在需要的时候进入系统	retrieval、vector store
Multi-agent	在单代理装不下时做职责拆分和上下文隔离	handoff、subagents、router