LangChain
源码原理与工程实践

从执行契约理解框架，沿真实调用链学习，并把可复用部分落实到工程实践。

读取日期：2026-10-04（中国标准时间）。LangChain 源码固定到 57236d55d9ffc8634bda987861e6fce460aff268，提交时间 2026-10-02 19:35:20 UTC。版本号来自该提交的包配置，不以网页上的动态版本替代。

配置依据 主包配置 · Core 配置 · Classic 配置 · 仓库结构说明

这张分层图用于定位职责，不表示所有请求都必须经过 Agent：一个 prompt | model | parser 链可直接使用 Core 和适配包。Agent 工厂依赖 Core 的数据契约，再把循环交给 LangGraph 执行。

源码定位 create_agent · BaseChatModel · BaseTool

Runnable[Input, Output] 把可执行组件统一成“输入 → 输出”契约。提示词、模型、解析器、检索器和工具都可以参与组合。LCEL 是这组组合能力的表达方式，其中 | 构造 RunnableSequence，字典等并行映射可转换为 RunnableParallel。

from langchain_core.runnables import RunnableLambda, RunnableParallel

add_one = RunnableLambda(lambda x: x + 1)
times_two = RunnableLambda(lambda x: x * 2)
chain = add_one | times_two

assert chain.invoke(3) == 8
assert chain.batch([1, 2]) == [4, 6]

fanout = RunnableParallel(original=lambda x: x, doubled=times_two)
assert fanout.invoke(3) == {"original": 3, "doubled": 6}

源码定位 Runnable · RunnableSequence · RunnableParallel · coerce_to_runnable

消息既承载对话内容，也记录工具调用与观察结果。把所有消息压成一个字符串，会丢失角色、工具调用 ID、多模态块和供应商元数据。

from langchain_core.messages import AIMessage, ToolMessage

action = AIMessage(
    content="",
    tool_calls=[{
        "name": "lookup_price",
        "args": {"sku": "A100"},
        "id": "call_01",
        "type": "tool_call",
    }],
)
observation = ToolMessage(
    content='{"price": 99.0, "currency": "CNY"}',
    name="lookup_price",
    tool_call_id="call_01",
    status="success",
)

源码定位 AIMessage · AIMessageChunk · ToolMessage · _patch_invalid_tool_calls

BaseChatModel.invoke() 把字符串、消息列表或 PromptValue 转换为统一输入，进入生成与 callback 链路，最终返回 AIMessage。具体适配器实现 _generate、_stream 及对应异步路径，把供应商 SDK 响应转换为框架消息。

OpenAI 适配器有消息转换、payload 构造、流 delta 转换等明确职责。应用逻辑应尽量使用标准字段；不得假定每个供应商都会填充完全相同的 response_metadata 或 reasoning 块。

源码定位 BaseChatModel.invoke · BaseChatModel._generate · init_chat_model · _convert_message_to_dict · BaseChatOpenAI._get_request_payload

@tool 将有类型签名和说明的函数转换为工具；BaseTool 提供名称、描述、参数 schema、调用、结果包装与错误配置。模型负责提出动作，运行时负责验证并执行动作。

from pydantic import BaseModel, Field
from langchain_core.tools import tool

class PriceArgs(BaseModel):
    sku: str = Field(min_length=1, description="商品 SKU")
    quantity: int = Field(ge=1, le=100, description="采购数量")

@tool(args_schema=PriceArgs)
def lookup_price(sku: str, quantity: int) -> str:
    """查询商品报价，返回单价和采购总价；此示例使用固定数据。"""
    prices = {"A100": 99.0, "B200": 129.0}
    if sku not in prices:
        return "未找到该 SKU；请检查商品编号。"
    unit_price = prices[sku]
    return f"SKU={sku}; 单价={unit_price}; 总价={unit_price * quantity}"

print(lookup_price.invoke({"sku": "A100", "quantity": 2}))

源码定位 tool · BaseTool.tool_call_schema · BaseTool._parse_input · BaseTool.run · ToolRuntime

工厂在创建时规范化模型、system prompt、工具、响应策略与中间件；合并状态 schema；注册 model、tools 和 hook 节点；连接条件边；最后 compile() 并返回可调用的图。循环发生在运行阶段，构图发生在创建阶段。

源码定位 create_agent · _build_commands · _make_model_to_tools_edge

当前工厂为每个待执行调用构造 Send("tools", [tool_call])。调用 ID 已有对应 ToolMessage 的动作会被排除；用于结构化输出的“工具”也会从普通业务工具分派中排除。

源码定位 _make_model_to_tools_edge · _make_tools_to_model_edge · create_agent · ModelCallLimitMiddleware · test_return_direct_graph.py

from langchain.agents.middleware import AgentMiddleware

class StableModelSettings(AgentMiddleware):
    def wrap_model_call(self, request, handler):
        updated = request.override(
            model_settings={**request.model_settings, "temperature": 0}
        )
        return handler(updated)

    async def awrap_model_call(self, request, handler):
        updated = request.override(
            model_settings={**request.model_settings, "temperature": 0}
        )
        return await handler(updated)

教学示例：模型须支持 temperature。request.override() 表达一次请求的覆盖，避免把共享模型实例直接改成某个用户的配置。

源码定位 AgentMiddleware · ModelRequest.override · _chain_model_call_handlers · _chain_tool_call_wrappers · test_composition.py

Agent 的 response_format 定义最终响应 schema。工厂将原始 schema 包成 AutoStrategy，在模型执行时判断使用供应商原生 schema 能力还是 ToolStrategy。自动选择会参考模型 profile 与兼容规则；显式传入策略则保留该选择。

import os
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.agents.middleware import ModelCallLimitMiddleware

class ReviewFinding(BaseModel):
    rule_id: str
    evidence: str = Field(description="与规则判断相关的文档证据")
    conclusion: str

agent = create_agent(
    model=ChatOpenAI(model=os.environ["LC_MODEL_ID"]),
    tools=[],
    response_format=ToolStrategy(ReviewFinding),
    middleware=[ModelCallLimitMiddleware(run_limit=4, exit_behavior="error")],
)
result = agent.invoke({
    "messages": [{"role": "user", "content": "规则 R1 要求签字；片段：签字页为空。"}]
})
finding: ReviewFinding = result["structured_response"]
print(finding.model_dump())

教学示例，需要凭据与支持 tool calling 的模型。 schema 校验保证字段形状；证据是否支持结论仍需业务规则与评估。

源码定位 ToolStrategy · ProviderStrategy · _supports_provider_strategy · _handle_structured_output_error · JsonOutputParser · test_response_format.py

from langchain.agents.middleware import ToolErrorMiddleware

def business_error(exc, request):
    if isinstance(exc, TimeoutError):
        return "查询暂时超时；可稍后重试或明确告诉用户当前不可用。"
    return None  # 未选择处理的异常继续传播

error_policy = ToolErrorMiddleware(on_error=business_error)

源码定位 _default_handle_tool_errors · ToolNode._execute_tool_sync · ToolErrorMiddleware · RunnableRetry · test_invalid_tool_calls.py

流式能力分三层：模型 token/chunk、Agent 的节点与状态进展、应用自定义进度。不要把工具参数的增量片段、工具执行事件与最终文本拼成同一条无类型字符串。

# 复用已创建的 agent；一个完整执行只调用一次 stream_events
run = agent.stream_events(
    {"messages": [{"role": "user", "content": "查询 A100 的报价"}]},
    version="v3",
)

for kind, item in run.interleave("messages", "tool_calls"):
    if kind == "messages":
        for delta in item.text:
            print(delta, end="", flush=True)
    elif kind == "tool_calls":
        print(f"\n执行工具：{item.tool_name}")
        for delta in item.output_deltas:
            print(delta, end="", flush=True)
        print(f"\n工具结果：{item.output}")

final_state = run.output

v3 的 message.tool_calls 展示模型生成的调用参数，run.tool_calls 展示实际执行生命周期。两个投影表达不同阶段。公开 reasoning 投影只有在模型确实返回相关内容时才有数据。

接口依据 官方 Event Streaming · 传统流模式 · Runnable.astream_events · create_agent

# 创建 agent 时传入 checkpointer=InMemorySaver()
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
# 应用应确认该 conversation_id 属于当前用户
config = {"configurable": {"thread_id": "conversation_001"}}

# 使用同一个带 checkpointer 的 agent，同一个 thread_id
# 第一轮：agent.invoke({"messages": [...]}, config=config)
# 下一轮只提供新消息，不重新拼接已保存的完整历史

源码定位 AgentState · add_messages · create_agent · SummarizationMiddleware · RunnableWithMessageHistory

RAG 把检索到的外部材料放入模型上下文，主要解决知识来源和证据引用。LangChain 的价值在于连接文档、切分、embedding、向量存储和检索器；召回质量、权限与结论可靠性需要另外评估。

from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter

documents = [Document(
    page_content="采购文件第一章……资格要求……报价规则……",
    metadata={"source": "tender-001", "page": 1, "version": "v1"},
)]
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=120,
    separators=["\n\n", "\n", "。", "；", "，", " ", ""],
    add_start_index=True,
)
chunks = splitter.split_documents(documents)
print(chunks[0].metadata)

源码定位 Document · RecursiveCharacterTextSplitter._split_text · TextSplitter._merge_splits · Embeddings · BaseRetriever · VectorStoreRetriever

RunnableConfig 用于运行维度的控制与追踪。链和模型把 tags、metadata、callbacks 等传给内部执行，形成父子运行关系。模型 API 参数应通过模型设置或 bind 传入；运行配置用于表达任务与执行环境。

源码定位 RunnableConfig · CallbackManager · BaseChatModel.invoke

源码定位 HumanInTheLoopMiddleware · ModelCallLimitMiddleware · ToolRuntime · RunnableRetry

阅读实现后，应回到 tests 观察框架如何定义行为。重点看状态与消息断言、调用顺序、出错后的路径和预算上限，而非只确认“最后生成了文字”。

# 在已固定版本的仓库中，先准备当前包环境
cd libs/langchain_v1
uv sync --group test

# 优先使用快速测试目标，避免完整测试启动数据库服务
make test_fast TEST_FILE=tests/unit_tests/agents/test_invalid_tool_calls.py

# 继续按同样方式选择具体用例文件
make test_fast TEST_FILE=tests/unit_tests/agents/middleware/core/test_composition.py

该提交的 make test 会启动 Postgres/Redis 服务并涉及 snapshot 更新；test_fast 适合先读和验证局部契约。不要直接把根目录当成统一 Python workspace，包级命令应在对应目录执行。

验证范围：检查了上述源码与测试定义；没有在本次环境执行 LangChain 完整测试套件或真实供应商调用。示例代码已做语法检查，HTML 的交互与布局另行验收。

建议先验证数据与执行契约，再接入外部模型。下面两个示例分别用于理解 LCEL 的固定数据流与 Agent 的工具循环。

from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.language_models.fake_chat_models import FakeListChatModel

class DemoRetriever(BaseRetriever):
    documents: list[Document]

    def _get_relevant_documents(self, query, *, run_manager):
        return self.documents

retriever = DemoRetriever(documents=[Document(
    page_content="供应商需在报价单上签字。",
    metadata={"source_id": "R1", "page": 3},
)])

def format_docs(docs):
    return "\n".join(
        f"[{d.metadata['source_id']}] 第{d.metadata['page']}页：{d.page_content}"
        for d in docs
    )

prompt = ChatPromptTemplate.from_messages([
    ("system", "只根据证据回答，保留证据 ID。"),
    ("human", "证据：\n{context}\n问题：{question}"),
])
model = FakeListChatModel(responses=["报价单需要签字 [R1]。"])
chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

answer = chain.invoke("报价单是否需要签字？")
assert answer == "报价单需要签字 [R1]。"
print(answer)

import os
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langchain_core.tools import tool

@tool
def lookup_policy(rule_id: str) -> str:
    """查询采购规则，输入规则 ID，返回带规则 ID 的证据文本。"""
    policies = {"R1": "报价单需签字，依据：[R1]。"}
    return policies.get(rule_id, "未找到该规则，证据不足。")

model = ChatOpenAI(
    model=os.environ["LC_MODEL_ID"],
    timeout=20,
    max_retries=1,
)
agent = create_agent(
    model=model,
    tools=[lookup_policy],
    system_prompt="查询规则后回答，保留证据 ID；无法查询时明确说明。",
    middleware=[ModelCallLimitMiddleware(run_limit=4, exit_behavior="error")],
)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "请查询 R1，说明报价单要求。"}]},
    config={"recursion_limit": 30, "tags": ["policy-demo"]},
)
print(result["messages"][-1].content)
for message in result["messages"]:
    print(type(message).__name__, getattr(message, "tool_call_id", None))

模型单次 timeout、SDK max_retries、图 recursion_limit 与模型阶段上限表达不同约束。实验 B 未设置整个业务任务的总 deadline，生产实现需要在调用入口补齐。

源码定位 BaseRetriever · ChatPromptTemplate · RunnableParallel · create_agent

以下是设计建议，用于自研 Java Agent Runtime。目标是迁移契约与执行语义；没有把 Python 类和完整 LangGraph 图引擎逐一照搬。

// Java 最小闭环示意，不是 LangChain 源码逐行翻译
while (budget.allowNextModelCall() && !context.isCancelled()) {
    ModelResponse response = model.generate(
        messageMapper.toModelInput(state.messages()), tools.schemas(), context
    );
    state.apply(response.toStateUpdate());

    List<ToolCall> pending = pendingCalls(response, state);
    if (pending.isEmpty()) {
        return completed(state);
    }

    for (ToolCall call : pending) {
        context.checkDeadlineAndCancellation();
        ToolResult result = executor.executeValidated(call, context);
        state.apply(result.toStateUpdate());
    }
}
return stopped(state, budgetOrCancellationReason(context));

用解释能力检查学习结果

所有 LangChain 源码链接固定到 57236d55d9ffc8634bda987861e6fce460aff268。作为底层机制补充的 LangGraph 源码单独固定到 9a0394d88b2211f299dcd69df92db3480c69ee61；它是独立仓库快照，不宣称与本次主包 uv.lock 解析出的分发版本完全相同。

1. Agent 工厂与路由
   libs/langchain_v1/langchain/agents/factory.py
2. 中间件类型与请求契约
   libs/langchain_v1/langchain/agents/middleware/types.py
3. 结构化输出策略
   libs/langchain_v1/langchain/agents/structured_output.py
4. Runnable 与组合算子
   libs/core/langchain_core/runnables/base.py
5. 运行配置
   libs/core/langchain_core/runnables/config.py
6. ChatModel 核心执行
   libs/core/langchain_core/language_models/chat_models.py
7. AI 消息与增量聚合
   libs/core/langchain_core/messages/ai.py
8. 工具结果消息
   libs/core/langchain_core/messages/tool.py
9. 工具输入与执行契约
   libs/core/langchain_core/tools/base.py
10. 函数到工具的转换
    libs/core/langchain_core/tools/convert.py
11. OpenAI 供应商适配
    libs/partners/openai/langchain_openai/chat_models/base.py
12. Retriever 契约
    libs/core/langchain_core/retrievers.py
13. VectorStore 与检索包装
    libs/core/langchain_core/vectorstores/base.py
14. 递归切分器
    libs/text-splitters/langchain_text_splitters/character.py
15. 工具执行错误转换
    libs/langchain_v1/langchain/agents/middleware/tool_error.py
16. 模型阶段调用上限
    libs/langchain_v1/langchain/agents/middleware/model_call_limit.py
17. 人工批准中间件
    libs/langchain_v1/langchain/agents/middleware/human_in_the_loop.py
18. LangGraph：工具节点与 runtime
    libs/prebuilt/langgraph/prebuilt/tool_node.py
19. LangGraph：消息 reducer
    libs/langgraph/langgraph/graph/message.py