LangChain入门：从 Runnable 协议到完整生态
标签: LangChain, 源码分析, Python, 架构设计, LLM
深入分析 LangChain 核心源码，从 Runnable 协议开始，逐步解析语言模型、工具系统、提示词模板、回调系统、记忆系统和检索器的内部实现机制
LangChain 核心源码导读
概述
本文档深入分析 LangChain 的核心源码，帮助初学者理解框架的内部实现机制。我们将从最基础的 Runnable 接口开始，逐步深入到各个核心模块。
源码结构导航
libs/
├── core/langchain_core/ # 核心抽象层
│ ├── runnables/ # ★ 核心：Runnable 协议
│ ├── language_models/ # 语言模型抽象
│ ├── messages/ # 消息类型
│ ├── tools/ # 工具抽象
│ ├── prompts/ # 提示词模板
│ ├── retrievers/ # 检索器抽象
│ ├── embeddings/ # 嵌入抽象
│ └── callbacks/ # 回调系统
│
├── langchain/langchain/ # 主实现
│ ├── chains/ # 链实现
│ ├── agents/ # 代理实现
│ ├── tools/ # 工具实现
│ ├── memory/ # 记忆管理
│ └── document_loaders/ # 文档加载器
│
└── community/ # 社区集成
└── langchain_community/
├── chat_models/ # 聊天模型
├── embeddings/ # 嵌入模型
├── vectorstores/ # 向量存储
└── tools/ # 社区工具
1. Runnable 协议 – 核心抽象
1.1 基础接口定义
文件: libs/core/langchain_core/runnables/base.py
from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Optional, Any
# 定义输入和输出的类型变量
Input = TypeVar("Input")
Output = TypeVar("Output")
class Runnable(Generic[Input, Output], ABC):
"""
LangChain 的核心抽象接口
所有组件都实现这个接口，提供统一的执行方式：
– invoke: 同步执行
– batch: 批量执行
– stream: 流式输出
"""
# ========== 核心同步方法 ==========
@abstractmethod
def invoke(
self,
input: Input,
config: Optional["RunnableConfig"] = None,
**kwargs: Any,
) –> Output:
"""同步执行：将输入转换为输出"""
raise NotImplementedError()
def batch(
self,
inputs: list[Input],
config: Optional["RunnableConfig"] = None,
**kwargs: Any,
) –> list[Output]:
"""
批量执行：处理多个输入
默认实现：使用线程池并行调用 invoke
子类可以重写以优化批量处理
"""
# 获取配置中的并发限制
config = config or {}
max_concurrency = kwargs.get("max_concurrency")
# 使用线程池并行执行
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
futures = [
executor.submit(self.invoke, inp, config)
for inp in inputs
]
return [f.result() for f in futures]
def stream(
self,
input: Input,
config: Optional["RunnableConfig"] = None,
**kwargs: Any,
) –> Iterator[Output]:
"""
流式输出：逐步产生结果
默认实现：调用 invoke 并产生单个结果
子类应该重写以支持真正的流式输出
"""
yield self.invoke(input, config, **kwargs)
# ========== 核心异步方法 ==========
async def ainvoke(
self,
input: Input,
config: Optional["RunnableConfig"] = None,
**kwargs: Any,
) –> Output:
"""
异步执行
默认实现：在线程池中运行 invoke
子类应该重写以提供原生异步实现
"""
from asyncio import get_event_loop
return await get_event_loop().run_in_executor(
None, functools.partial(self.invoke, input, config, **kwargs)
)
async def abatch(
self,
inputs: list[Input],
config: Optional["RunnableConfig"] = None,
**kwargs: Any,
) –> list[Output]:
"""异步批量执行"""
import asyncio
# 并行执行所有异步调用
coros = [self.ainvoke(inp, config, **kwargs) for inp in inputs]
return await asyncio.gather(*coros)
async def astream(
self,
input: Input,
config: Optional["RunnableConfig"] = None,
**kwargs: Any,
) –> AsyncIterator[Output]:
"""异步流式输出"""
result = await self.ainvoke(input, config, **kwargs)
yield result
# ========== 组合操作 ==========
def __or__(
self,
other: "Runnable[Any, Output]",
) –> "RunnableSequence[Input, Any]":
"""
实现 | 操作符：创建顺序链
这是 LCEL (LangChain Expression Language) 的核心
"""
from langchain_core.runnables.base import RunnableSequence
return RunnableSequence([self, other])
def __ror__(
self,
other: "Runnable[Input, Any]",
) –> "RunnableSequence[Input, Output]":
"""反向 | 操作符"""
from langchain_core.runnables.base import RunnableSequence
return RunnableSequence([other, self])
1.2 RunnableSequence – 顺序链
文件: libs/core/langchain_core/runnables/base.py
class RunnableSequence(RunnableSerializable[Input, Output]):
"""
顺序执行的链：LCEL | 操作符的核心实现
当使用 | 操作符时，实际创建的就是 RunnableSequence
"""
steps: Sequence[Runnable[Any, Any]]
"""要按顺序执行的 Runnable 列表"""
def invoke(
self,
input: Input,
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) –> Output:
"""
顺序执行所有步骤
执行流程：
1. 将输入传递给第一个步骤
2. 每个步骤的输出成为下一个步骤的输入
3. 返回最后一个步骤的输出
"""
value: Any = input
# 依次执行每个步骤
for i, step in enumerate(self.steps):
try:
# 将当前值传递给步骤
value = step.invoke(value, config, **kwargs)
except Exception as e:
# 添加上下文信息以便调试
raise RuntimeError(
f"Error in step {i} ({step.get_name()}): {str(e)}"
) from e
return value
async def ainvoke(
self,
input: Input,
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) –> Output:
"""异步版本"""
value: Any = input
for i, step in enumerate(self.steps):
try:
value = await step.ainvoke(value, config, **kwargs)
except Exception as e:
raise RuntimeError(
f"Error in async step {i} ({step.get_name()}): {str(e)}"
) from e
return value
def stream(
self,
input: Input,
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) –> Iterator[Output]:
"""
流式执行
关键点：最后一个步骤可以流式输出时，
整个链就可以流式输出
"""
value: Any = input
# 前面的步骤正常执行
for i, step in enumerate(self.steps[:–1]):
value = step.invoke(value, config, **kwargs)
# 最后一个步骤流式输出
for chunk in self.steps[–1].stream(value, config, **kwargs):
yield chunk
使用示例:
# 这段代码创建了一个 RunnableSequence
chain = prompt | llm | parser
# 等价于：
chain = RunnableSequence([prompt, llm, parser])
# 执行时的数据流：
# input -> prompt.format() -> llm.invoke() -> parser.parse() -> output
1.3 RunnableParallel – 并行执行
文件: libs/core/langchain_core/runnables/base.py
class RunnableParallel(RunnableSerializable[Input, dict[str, Any]]):
"""
并行执行多个 Runnable
可以使用字典字面量语法创建：
chain = {
"summary": summary_chain,
"analysis": analysis_chain,
}
"""
steps: Mapping[str, Runnable[Any, Any]]
"""要并行执行的 Runnable 映射"""
def invoke(
self,
input: Input,
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) –> dict[str, Any]:
"""
并行执行所有步骤
执行流程：
1. 将相同输入传递给所有步骤
2. 并行执行所有步骤
3. 将结果合并到字典中返回
"""
import asyncio
# 创建异步任务
async def _parallel_invoke():
results = {}
tasks = []
# 为每个步骤创建异步任务
for key, step in self.steps.items():
async def invoke_step(k=key, s=step):
result = await s.ainvoke(input, config, **kwargs)
return k, result
tasks.append(invoke_step())
# 等待所有任务完成
completed = await asyncio.gather(*tasks)
return {k: v for k, v in completed}
# 在同步上下文中运行异步代码
return asyncio.run(_parallel_invoke())
1.4 RunnableLambda – 函数适配器
文件: libs/core/langchain_core/runnables/base.py
class RunnableLambda(Runnable[Input, Output]):
"""
将普通函数转换为 Runnable
这使得任何函数都可以在 LCEL 链中使用
"""
func: Union[
Callable[[Input], Output], # 同步函数
Callable[[Input], Awaitable[Output]], # 异步函数
]
"""要包装的函数"""
def invoke(
self,
input: Input,
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) –> Output:
"""调用包装的函数"""
return self.func(input)
async def ainvoke(
self,
input: Input,
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) –> Output:
"""异步调用包装的函数"""
result = self.func(input)
# 如果结果是可等待对象，等待它
if inspect.isawaitable(result):
return await result
return result
使用示例:
# 将普通函数转换为 Runnable
def extract_title(text: str) –> str:
"""提取文本的第一行作为标题"""
return text.split('\\n')[0]
# 现在可以在链中使用
chain = (
RunnableLambda(extract_title)
| another_runnable
)
1.5 RunnablePassthrough – 透传
文件: libs/core/langchain_core/runnables/passthrough.py
class RunnablePassthrough(RunnableSerializable[Other, Other]):
"""
透传输入：将输入原样传递，不进行任何修改
常用于 RunnableParallel 中，需要同时传递输入和使用处理后的结果
"""
def invoke(
self,
input: Other,
config: Optional[RunnableConfig] = None,
) –> Other:
"""直接返回输入"""
return input
使用示例:
from langchain_core.runnables import RunnablePassthrough
# 在并行链中同时保留原始输入
chain = {
"original": RunnablePassthrough(), # 保留原始输入
"processed": some_processor, # 处理后的结果
"count": lambda x: len(x), # 计数
}
2. 语言模型抽象
2.1 BaseLanguageModel
文件: libs/core/langchain_core/language_models/base.py
class BaseLanguageModel(
RunnableSerializable[LanguageModelInput, LanguageModelOutputVar], ABC
):
"""
所有语言模型的抽象基类
提供了生成文本的核心接口，所有具体的 LLM 实现都继承此类
"""
# 配置字段
cache: Union[BaseCache, bool, None] = Field(default=None, exclude=True)
"""响应缓存"""
verbose: bool = Field(default_factory=_get_verbosity, exclude=True)
"""是否打印输出"""
callbacks: Callbacks = Field(default=None, exclude=True)
"""回调处理器"""
tags: Optional[list[str]] = Field(default=None, exclude=True)
"""用于追踪的标签"""
metadata: Optional[dict[str, Any]] = Field(default=None, exclude=True)
"""元数据"""
@abstractmethod
def generate_prompt(
self,
prompts: list[PromptValue],
stop: list[str] | None = None,
**kwargs: Any,
) –> LLMResult:
"""
从提示生成文本
这是核心抽象方法，子类必须实现
"""
raise NotImplementedError()
def invoke(
self,
input: LanguageModelInput,
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) –> LanguageModelOutputVar:
"""
Runnable 接口的实现
将输入转换为 PromptValue，然后调用 generate_prompt
"""
# 转换输入为 PromptValue
if isinstance(input, str):
prompt_value = StringPromptValue(text=input)
elif isinstance(input, PromptValue):
prompt_value = input
else:
# 消息列表
prompt_value = ChatPromptValue(messages=input)
# 调用 generate_prompt
llm_result = self.generate_prompt(
[prompt_value],
**kwargs
)
# 提取结果
return llm_result.generations[0][0].text
2.2 ChatModel – 聊天模型
文件: libs/core/langchain_core/language_models/chat_models.py
class BaseChatModel(BaseLanguageModel[BaseMessage], ABC):
"""
聊天模型的抽象基类
处理消息列表而不是纯文本
"""
@abstractmethod
def _generate(
self,
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: CallbackManagerForLLMRun | None = None,
**kwargs: Any,
) –> ChatResult:
"""
生成聊天响应
子类必须实现此方法
"""
raise NotImplementedError()
def bind_tools(
self,
tools: Sequence[BaseTool | StructuredTool],
**kwargs: Any,
) –> "RunnableBinding":
"""
绑定工具到模型
这使得模型可以调用工具
"""
# 将工具转换为模型特定的格式
formatted_tools = [self._format_tool(tool) for tool in tools]
# 创建绑定
return RunnableBinding(
bound=self,
kwargs={**kwargs, "tools": formatted_tools}
)
3. 工具系统
3.1 BaseTool
文件: libs/core/langchain_core/tools/base.py
class BaseTool(RunnableSerializable[Union[str, dict], Any], ABC):
"""
工具的抽象基类
工具是可以被 LLM 调用的函数，用于执行特定任务
"""
# 工具元数据
name: str
"""工具名称，在提示词中使用"""
description: str
"""工具描述，告诉 LLM 何时使用此工具"""
args_schema: Type[BaseModel] = Field(default=...)
"""参数的 Pydantic 模式，用于验证和文档"""
# ========== 核心方法 ==========
@abstractmethod
def _run(
self,
*args: Any,
**kwargs: Any,
) –> Any:
"""
同步执行工具
子类必须实现此方法
"""
raise NotImplementedError()
async def _arun(
self,
*args: Any,
**kwargs: Any,
) –> Any:
"""
异步执行工具
默认实现调用 _run
子类可以重写以提供原生异步实现
"""
return await asyncio.get_event_loop().run_in_executor(
None, functools.partial(self._run, *args, **kwargs)
)
# ========== Runnable 接口实现 ==========
def invoke(
self,
input: Union[str, dict],
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) –> Any:
"""
Runnable 接口实现
解析输入并执行工具
"""
# 解析输入
if isinstance(input, str):
parsed_input = {"query": input}
elif isinstance(input, dict):
parsed_input = input
else:
parsed_input = self.args_schema.model_validate(input)
# 执行工具
return self._run(**parsed_input.model_dump())
# ========== 工具特定方法 ==========
def to_tool_call(
self,
arguments: dict[str, Any],
) –> ToolCall:
"""创建工具调用记录"""
return ToolCall(
name=self.name,
args=arguments,
id=str(uuid.uuid4()),
)
3.2 StructuredTool – 结构化工具
文件: libs/core/langchain_core/tools/structured.py
class StructuredTool(BaseTool):
"""
带有结构化输入/输出的工具
允许为工具的输入和输出定义严格的模式
"""
args_schema: Type[BaseModel] = Field(default=...)
"""输入模式"""
response_format: "Type[BaseModel]" = Field(default=None)
"""输出模式（可选）"""
func: Callable[..., Any]
"""要包装的函数"""
def _run(
self,
*args: Any,
**kwargs: Any,
) –> Any:
"""调用包装的函数"""
return self.func(*args, **kwargs)
@classmethod
def from_function(
cls,
func: Callable,
name: str,
description: str,
args_schema: Optional[Type[BaseModel]] = None,
) –> "StructuredTool":
"""
从函数创建工具
这是创建自定义工具的便捷方法
"""
# 如果没有提供模式，从函数签名推断
if args_schema is None:
args_schema = cls._infer_schema(func)
return cls(
name=name,
description=description,
args_schema=args_schema,
func=func,
)
创建自定义工具示例:
from langchain_core.tools import StructuredTool
from pydantic import BaseModel
class SearchInput(BaseModel):
"""搜索输入模式"""
query: str
limit: int = 10
def search_func(query: str, limit: int = 10) –> str:
"""执行搜索"""
return f"Searching for: {query}"
# 创建工具
search_tool = StructuredTool.from_function(
func=search_func,
name="search",
description="搜索互联网",
args_schema=SearchInput,
)
4. 提示词模板
4.1 BasePromptTemplate
文件: libs/core/langchain_core/prompts/base.py
class BasePromptTemplate(Serializable, ABC):
"""
提示词模板的抽象基类
定义了模板的接口和基本行为
"""
template: str
"""模板字符串"""
input_variables: list[str]
"""模板中使用的变量名"""
template_format: str = "f-string"
"""模板格式：f-string, jinja2, mustache"""
@abstractmethod
def format(self, **kwargs: Any) –> str:
"""
格式化模板
使用提供的值替换模板中的变量
"""
raise NotImplementedError()
def invoke(
self,
input: dict[str, Any],
config: Optional[RunnableConfig] = None,
) –> PromptValue:
"""Runnable 接口实现"""
return self.format_prompt(**input)
def __or__(
self,
other: Runnable[Any, Other],
) –> RunnableSequence[dict[str, Any], Other]:
"""支持 | 操作符"""
return RunnableSequence([self, other])
4.2 ChatPromptTemplate
文件: libs/core/langchain_core/prompts/chat.py
class ChatPromptTemplate(BasePromptTemplate):
"""
聊天提示词模板
由多个消息模板组成
"""
messages: Sequence[MessageLikeRepresentation]
"""消息模板列表"""
@classmethod
def from_template(
cls,
template: str,
**kwargs: Any,
) –> "ChatPromptTemplate":
"""
从字符串创建聊天模板
这是创建聊天提示词的便捷方法
"""
return cls(
messages=[
("human", template) # 人类消息
],
**kwargs
)
@classmethod
def from_messages(
cls,
messages: Sequence[MessageLikeRepresentation],
) –> "ChatPromptTemplate":
"""
从消息列表创建模板
消息可以是：
– 元组：(角色, 内容)
– MessageTemplate：消息模板对象
– BaseMessage：实际消息
"""
return cls(messages=list(messages))
def format_messages(
self,
**kwargs: Any,
) –> list[BaseMessage]:
"""格式化为消息列表"""
formatted_messages = []
for message in self.messages:
if isinstance(message, tuple):
role, content = message
# 格式化内容
formatted_content = content.format(**kwargs)
# 创建对应的消息类型
if role == "system":
formatted_messages.append(SystemMessage(content=formatted_content))
elif role == "human":
formatted_messages.append(HumanMessage(content=formatted_content))
elif role == "ai":
formatted_messages.append(AIMessage(content=formatted_content))
else:
formatted_messages.append(message)
return formatted_messages
5. 回调系统
5.1 BaseCallbackHandler
文件: libs/core/langchain_core/callbacks/base.py
class BaseCallbackHandler(ABC):
"""
回调处理器的抽象基类
定义了可以监听的各种事件
"""
# ========== LLM 事件 ==========
def on_llm_start(
self,
prompts: list[str],
**kwargs: Any,
) –> Any:
"""LLM 开始生成"""
pass
def on_llm_end(
self,
response: LLMResult,
**kwargs: Any,
) –> Any:
"""LLM 结束生成"""
pass
def on_llm_error(
self,
error: Exception,
**kwargs: Any,
) –> Any:
"""LLM 发生错误"""
pass
# ========== 链事件 ==========
def on_chain_start(
self,
serialized: dict,
inputs: dict,
**kwargs: Any,
) –> Any:
"""链开始执行"""
pass
def on_chain_end(
self,
outputs: dict,
**kwargs: Any,
) –> Any:
"""链结束执行"""
pass
def on_chain_error(
self,
error: Exception,
**kwargs: Any,
) –> Any:
"""链发生错误"""
pass
# ========== 工具事件 ==========
def on_tool_start(
self,
serialized: dict,
input_str: str,
**kwargs: Any,
) –> Any:
"""工具开始执行"""
pass
def on_tool_end(
self,
output: str,
**kwargs: Any,
) –> Any:
"""工具结束执行"""
pass
def on_tool_error(
self,
error: Exception,
**kwargs: Any,
) –> Any:
"""工具发生错误"""
pass
5.2 CallbackManager
文件: libs/core/langchain_core/callbacks/manager.py
class CallbackManager:
"""
管理多个回调处理器
当事件发生时，通知所有注册的处理器
"""
handlers: list[BaseCallbackHandler]
def __init__(self, handlers: list[BaseCallbackHandler]):
self.handlers = handlers
def on_llm_start(
self,
prompts: list[str],
**kwargs: Any,
) –> None:
"""通知所有处理器：LLM 开始"""
for handler in self.handlers:
try:
handler.on_llm_start(prompts, **kwargs)
except Exception as e:
logger.warning(f"Error in callback handler: {e}")
def on_llm_end(
self,
response: LLMResult,
**kwargs: Any,
) –> None:
"""通知所有处理器：LLM 结束"""
for handler in self.handlers:
try:
handler.on_llm_end(response, **kwargs)
except Exception as e:
logger.warning(f"Error in callback handler: {e}")
6. 记忆系统
6.1 BaseMemory
文件: libs/langchain/langchain/memory/base.py
class BaseMemory(Serializable, ABC):
"""
记忆的抽象基类
定义了在对话中保存和检索状态的接口
"""
@property
@abstractmethod
def memory_variables(self) –> list[str]:
"""
此记忆组件管理的变量名
例如：["history", "summary"]
"""
pass
@abstractmethod
def load_memory_variables(
self,
inputs: dict[str, Any],
) –> dict[str, Any]:
"""
加载记忆变量
返回的字典会被添加到提示词中
"""
pass
@abstractmethod
def save_context(
self,
inputs: dict[str, Any],
outputs: dict[str, Any],
) –> None:
"""
保存当前对话上下文到记忆
inputs: 用户的输入
outputs: 模型的输出
"""
pass
def clear(self) –> None:
"""清除所有记忆"""
pass
6.2 ConversationBufferMemory
文件: libs/langchain/langchain/memory/buffer.py
class ConversationBufferMemory(BaseMemory):
"""
缓冲对话历史的记忆组件
保存所有对话消息
"""
chat_memory: BaseChatMessageHistory = Field(default_factory=ChatMessageHistory)
"""消息历史存储"""
return_messages: bool = False
"""是否返回消息对象而非字符串"""
@property
def memory_variables(self) –> list[str]:
return ["history"]
def load_memory_variables(
self,
inputs: dict[str, Any],
) –> dict[str, Any]:
"""加载对话历史"""
if self.return_messages:
return {"history": self.chat_memory.messages}
else:
# 转换为字符串
return {
"history": get_buffer_string(
self.chat_memory.messages,
human_prefix="Human",
ai_prefix="AI",
)
}
def save_context(
self,
inputs: dict[str, Any],
outputs: dict[str, Any],
) –> None:
"""保存对话"""
# 保存用户输入
input_str = inputs.get("input", "")
self.chat_memory.add_message(HumanMessage(content=input_str))
# 保存模型输出
output_str = outputs.get("output", "")
self.chat_memory.add_message(AIMessage(content=output_str))
7. 检索器
7.1 BaseRetriever
文件: libs/core/langchain_core/retrievers.py
class BaseRetriever(Runnable[langchain_core.utils.input.Input, list[Document]], ABC):
"""
检索器的抽象基类
检索器用于根据查询找到相关文档
"""
@abstractmethod
def _get_relevant_documents(
self,
query: str,
run_manager: CallbackManagerForChainRun,
) –> list[Document]:
"""
获取相关文档
子类必须实现此方法
"""
raise NotImplementedError()
def invoke(
self,
input: str,
config: Optional[RunnableConfig] = None,
) –> list[Document]:
"""Runnable 接口实现"""
# 创建运行管理器
run_manager = get_callback_manager_for_config(config)
# 调用子类实现
return self._get_relevant_documents(input, run_manager)
async def ainvoke(
self,
input: str,
config: Optional[RunnableConfig] = None,
) –> list[Document]:
"""异步版本"""
run_manager = await get_async_callback_manager_for_config(config)
return await self._aget_relevant_documents(input, run_manager)
7.2 VectorStoreRetriever
文件: libs/langchain/langchain/vectorstores/base.py
class VectorStoreRetriever(BaseRetriever):
"""
基于向量存储的检索器
使用向量相似度搜索查找相关文档
"""
vectorstore: VectorStore
"""底层的向量存储"""
search_type: str = "similarity"
"""搜索类型：similarity, mmr, similarity_score_threshold"""
search_kwargs: dict[str, Any] = Field(default_factory=dict)
"""搜索参数，如 k（返回数量）"""
def _get_relevant_documents(
self,
query: str,
run_manager: CallbackManagerForChainRun,
) –> list[Document]:
"""执行向量搜索"""
if self.search_type == "similarity":
# 标准相似度搜索
docs = self.vectorstore.similarity_search(
query,
**self.search_kwargs
)
elif self.search_type == "mmr":
# 最大边际相关性（MMR）搜索
docs = self.vectorstore.max_marginal_relevance_search(
query,
**self.search_kwargs
)
else:
raise ValueError(f"Unknown search type: {self.search_type}")
return docs
8. 数据流追踪
8.1 使用 LangSmith 追踪
LangChain 与 LangSmith 深度集成，可以追踪所有执行：
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
# 所有链执行都会被自动追踪
result = chain.invoke({"input": "hello"})
8.2 本地调试
from langchain_core.globals import set_debug
from langchain_core.tracers import ConsoleCallbackHandler
# 启用调试模式
set_debug(True)
# 或使用控制台回调
chain.invoke(
{"input": "hello"},
config={"callbacks": [ConsoleCallbackHandler()]}
)
9. 总结
LangChain 的核心设计理念：
通过理解这些核心源码，你可以：
- 更好地使用 LangChain
- 创建自定义组件
- 调试和优化你的应用
- 贡献到 LangChain 项目
