基于 langchain-ai/langchain 源码分析,覆盖 langchain-core 与 langchain 两个核心包
1. 架构哲学:可组合的 Runnable 抽象
LangChain 的核心设计哲学是 “一切皆 Runnable”。所有组件——LLM、Prompt、OutputParser、Tool、Retriever——都继承自同一个泛型基类 Runnable[Input, Output],从而获得统一的调用接口和组合能力。
class Runnable(ABC, Generic[Input, Output]):
"""A unit of work that can be invoked, batched, streamed, transformed and composed."""
name: str | None
"""The name of the `Runnable`. Used for debugging and tracing."""
@abstractmethod
def invoke(self, input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Output:
"""Transform a single input into an output."""
async def ainvoke(self, input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Output:
return await run_in_executor(config, self.invoke, input, config, **kwargs)
关键设计决策:
- LCEL(LangChain Expression Language):通过
|管道运算符实现声明式链式组合,RunnableSequence、RunnableParallel是核心组合原语 - 同步/异步双模:每个 Runnable 同时提供
invoke/ainvoke、stream/astream、batch/abatch,异步默认通过线程池桥接同步实现 - Pydantic 驱动的 Schema:输入/输出类型通过泛型参数自动推导,生成 JSON Schema 用于验证和文档
def __or__(self, other) -> RunnableSerializable[Input, Any]:
"""Compose this Runnable with another object to create a RunnableSequence."""
return RunnableSequence(self, coerce_to_runnable(other))
def pipe(self, *others, name=None) -> RunnableSerializable[Input, Other]:
"""Equivalent to RunnableSequence(self, *others) or self | others[0] | ..."""
return RunnableSequence(self, *others, name=name)
2. Agent Loop:观察-行动循环
LangChain 的 Agent 采用经典的 ReAct(Reasoning + Acting) 模式。核心循环定义在 langchain_core/agents.py 中,通过两个关键数据结构驱动:
class AgentAction(Serializable):
"""Represents a request to execute an action by an agent."""
tool: str # 要执行的工具名称
tool_input: str | dict[Any, Any] # 传递给工具的输入
log: str # LLM 的推理过程日志
type: Literal["AgentAction"] = "AgentAction"
@property
def messages(self) -> Sequence[BaseMessage]:
"""Return the messages that correspond to this action."""
return _convert_agent_action_to_messages(self)
class AgentFinish(Serializable):
"""Agents return an AgentFinish when they have reached a stopping condition."""
return_values: dict[Any, Any] # 最终返回值
log: str # 完整的 LLM 预测文本
type: Literal["AgentFinish"] = "AgentFinish"
Agent Loop 的核心流程:
- LLM 根据 prompt 生成
AgentAction(选择工具和输入) - 执行工具,获得 observation
- 将 observation 转换为消息返回给 LLM
- LLM 决定下一步——要么生成新的
AgentAction,要么生成AgentFinish
def _convert_agent_observation_to_messages(agent_action, observation):
"""Convert agent observation to messages for the next iteration."""
if isinstance(observation, str):
content = observation
else:
try:
content = json.dumps(observation, ensure_ascii=False)
except Exception:
content = str(observation)
return [HumanMessage(content=content)]
3. 工具系统:BaseTool 与函数式工具
工具系统在 langchain_core/tools/base.py 中实现,BaseTool 同时继承 RunnableSerializable,使工具可以无缝嵌入 LCEL 链。
class BaseTool(RunnableSerializable[str | dict[str, Any] | ToolCall, Any]):
"""Base class for all LangChain tools."""
name: str
"""The unique name of the tool that clearly communicates its purpose."""
description: str
"""Used to tell the model how/when/why to use the tool."""
args_schema: Annotated[ArgsSchema | None, SkipValidation()] = Field(default=None)
"""Pydantic model class to validate and parse the tool's input arguments."""
return_direct: bool = False
"""Whether to return the tool's output directly. Setting True means
AgentExecutor will stop looping after this tool is called."""
handle_tool_error: bool | str | Callable | None = False
"""Handle ToolException raised by tool execution."""
response_format: Literal["content", "content_and_artifact"] = "content"
"""The tool response format."""
工具定义支持多种方式:
- 继承
BaseTool实现_run/_arun - 使用
@tool装饰器从函数自动生成 Schema - 支持
InjectedToolArg注入运行时参数(不暴露给 LLM)
def create_schema_from_function(model_name, func, *, filter_args=None, parse_docstring=False):
"""Create a Pydantic schema from a function's signature."""
sig = inspect.signature(func)
# 使用 Pydantic 的 validate_arguments 自动推导参数 Schema
validated = validate_arguments(func, config=_SchemaConfig)
inferred_model = validated.model
4. 流式输出:多层流式架构
LangChain 的流式设计分为三个层次:
第一层:基础 stream/astream
def stream(self, input: Input, config: RunnableConfig | None = None, **kwargs) -> Iterator[Output]:
"""Default implementation of stream, which calls invoke.
Subclasses must override this method if they support streaming output."""
yield self.invoke(input, config, **kwargs)
第二层:astream_events(结构化事件流)
def astream_events(self, input, config=None, *, version="v2", **kwargs) -> AsyncIterator[StreamEvent]:
"""Generate a stream of events.
A StreamEvent is a dictionary with:
- event: 'on_[runnable_type]_(start|stream|end)'
- name: The name of the Runnable
- run_id: Randomly generated ID
- parent_ids: IDs of parent runnables
- data: Event-specific data (chunk, input, output)
"""
第三层:astream_log(JSON Patch 差分流,已废弃)
async def astream_log(self, input, config=None, *, diff=True, **kwargs):
"""Stream all output from a Runnable as Log objects with Jsonpatch ops.
Deprecated since 1.3.3, use astream instead."""
warn_deprecated(since="1.3.3", message="astream_log is deprecated. Use astream instead.")
5. 错误处理:分层容错机制
LangChain 在工具层面实现了精细的错误处理策略:
# 工具级错误处理
handle_tool_error: bool | str | Callable[[ToolException], ToolExceptionHandlerOutput] | None = False
handle_validation_error: bool | str | Callable[[ValidationError], str] | None = False
# 错误处理逻辑
except (ValidationError, ValidationErrorV1) as e:
if not self.handle_validation_error:
error_to_raise = e
else:
content = _handle_validation_error(e, flag=self.handle_validation_error)
status = "error"
except ToolException as e:
if not self.handle_tool_error:
error_to_raise = e
else:
content = _handle_tool_error(e, flag=self.handle_tool_error)
status = "error"
Runnable 级别的重试和 Fallback:
# 通过 with_retry 装饰器添加重试策略
sequence = (
RunnableLambda(add_one) |
RunnableLambda(buggy_double).with_retry(
stop_after_attempt=10,
wait_exponential_jitter=False
)
)
# batch 中的 return_exceptions 模式
def batch(self, inputs, config=None, *, return_exceptions=False, **kwargs):
"""return_exceptions: Whether to return exceptions instead of raising them."""
def invoke(input_, config):
if return_exceptions:
try:
return self.invoke(input_, config, **kwargs)
except Exception as e:
return e
else:
return self.invoke(input_, config, **kwargs)
6. 上下文管理:RunnableConfig
所有 Runnable 方法都接受 RunnableConfig 参数,用于传递执行上下文:
class RunnableConfig(TypedDict, total=False):
tags: list[str] # 用于过滤和分类
metadata: dict[str, Any] # 追踪元数据
callbacks: Callbacks # 回调处理器
run_name: str # 运行名称
max_concurrency: int # 最大并发数
recursion_limit: int # 递归深度限制
configurable: dict[str, Any] # 可配置字段
可配置字段支持运行时动态替换组件:
def config_schema(self, *, include=None) -> type[BaseModel]:
"""The type of config this Runnable accepts specified as a Pydantic model.
To mark a field as configurable, see configurable_fields and
configurable_alternatives methods."""
config_specs = self.config_specs
configurable = create_model_v2("Configurable", field_definitions={
spec.id: (spec.annotation, Field(spec.default, title=spec.name))
for spec in config_specs
})
7. 记忆系统:消息体系
LangChain 定义了丰富的消息类型层次结构,支持多模态内容:
# 核心消息类型(langchain_core/messages/__init__.py)
__all__ = (
"AIMessage", "AIMessageChunk", # AI 回复
"HumanMessage", "HumanMessageChunk", # 用户输入
"SystemMessage", "SystemMessageChunk", # 系统提示
"ToolMessage", "ToolMessageChunk", # 工具返回
"FunctionMessage", "FunctionMessageChunk", # 函数调用
"ChatMessage", "ChatMessageChunk", # 自定义角色
"RemoveMessage", # 消息修改器
# 多模态内容块
"TextContentBlock", "ImageContentBlock", "AudioContentBlock",
"VideoContentBlock", "FileContentBlock", "DataContentBlock",
"ReasoningContentBlock", # 推理内容
# 工具调用
"ToolCall", "ToolCallChunk", "InvalidToolCall",
"ServerToolCall", "ServerToolResult",
# 消息操作工具函数
"filter_messages", "trim_messages", "merge_message_runs",
"convert_to_messages", "convert_to_openai_messages",
"get_buffer_string",
)
消息系统的关键工具函数:
filter_messages:按类型/名称/ID 过滤消息trim_messages:按 token 数裁剪上下文窗口merge_message_runs:合并连续的同类消息
8. 任务规划:图与链的组合
LangChain 通过图结构支持任务规划和可视化:
def get_graph(self, config: RunnableConfig | None = None) -> Graph:
"""Return a graph representation of this Runnable."""
graph = Graph()
input_node = graph.add_node(self.get_input_schema(config))
runnable_node = graph.add_node(self, metadata=config.get("metadata") if config else None)
output_node = graph.add_node(self.get_output_schema(config))
graph.add_edge(input_node, runnable_node)
graph.add_edge(runnable_node, output_node)
return graph
更复杂的 Agent 规划能力已迁移到 LangGraph,它提供了基于状态图的 Agent 编排框架,支持条件分支、循环、子图等高级模式。
9. 可观测性:回调系统与事件流
LangChain 的可观测性建立在回调系统之上:
# 所有 Runnable 都支持 callbacks 参数
callbacks: Callbacks = Field(default=None, exclude=True)
tags: list[str] | None = None
metadata: dict[str, Any] | None = None
# 调试模式
from langchain_core.globals import set_debug
set_debug(True) # 启用全局调试输出
# 或者使用 ConsoleCallbackHandler
from langchain_core.tracers import ConsoleCallbackHandler
chain.invoke(..., config={"callbacks": [ConsoleCallbackHandler()]})
事件流提供结构化的运行时信息:
# StreamEvent schema
{
"event": "on_chat_model_start", # 事件类型
"name": "model_name", # 组件名称
"run_id": "uuid", # 运行 ID
"parent_ids": [], # 父组件 ID 链
"tags": [], # 标签
"metadata": {}, # 元数据
"data": {"chunk": ...} # 事件数据
}
配套的 LangSmith 提供了生产级的追踪、评估和调试 UI。
10. 扩展机制:多层扩展点
LangChain 提供了丰富的扩展机制:
Schema 扩展: 自定义输入/输出 Schema,支持 ConfigurableField 运行时替换
@property
def config_specs(self) -> list[ConfigurableFieldSpec]:
"""List configurable fields for this Runnable."""
return []
组合扩展: 通过 RunnableLambda 包装任意函数
sequence = RunnableLambda(lambda x: x + 1) | RunnableLambda(lambda x: x * 2)
回调扩展: 自定义 BaseCallbackHandler 监听所有生命周期事件
工具扩展: 通过 @tool 装饰器或继承 BaseTool 创建自定义工具
集成扩展: 通过 langchain-community 包扩展模型、向量存储、检索器等
# 快速初始化不同提供商的模型
from langchain.chat_models import init_chat_model
model = init_chat_model("openai:gpt-5.5") # 通过 provider:model 格式切换
总结
| 维度 | 核心设计 | 关键类/接口 |
|---|
|——|———|————|
| 架构哲学 | 一切皆 Runnable,LCEL 管道组合 | Runnable, RunnableSequence |
|---|---|---|
| Agent Loop | ReAct 观察-行动循环 | AgentAction, AgentFinish |
| 工具系统 | BaseTool + 函数式工具 + Schema 自动推导 | BaseTool, @tool |
| 流式输出 | 三层流式:stream → astream_events → astream_log | StreamEvent |
| 错误处理 | 工具级 handle_tool_error + Runnable 级 retry/fallback | ToolException |
| 上下文管理 | RunnableConfig 统一配置传递 | RunnableConfig |
| 记忆系统 | 分层消息类型 + 多模态内容块 | BaseMessage 体系 |
| 任务规划 | Graph 可视化 + LangGraph 状态图编排 | Graph |
| 可观测性 | 回调系统 + 结构化事件流 + LangSmith | CallbackManager |
| 扩展机制 | Schema/组合/回调/工具/集成五层扩展 | ConfigurableField |
LangChain 的架构演进路径清晰:从早期的 Chain 类层次 → LCEL 管道抽象 → LangGraph 状态图编排。核心 langchain-core 保持轻量和稳定,复杂的 Agent 编排能力逐步迁移到 LangGraph 和 Deep Agents 等上层框架。
发表回复