Day16（补充）：LangChain 组件完全指南：核心函数

🌟 引言

LangChain 自诞生以来，已成为 LLM 应用开发的事实标准。2024-2025 年间，框架经历了翻天覆地的重构：LCEL（LangChain Expression Language） 成为核心，Runnable 接口统一了所有组件，旧式 LLMChain 被废弃，LangGraph 接管了复杂 Agent 的编排。

本文基于 LangChain v0.3+，精心整理出 122 个最常用、最核心的组件、函数和类。每个条目均包含：功能解释、用法要点、适用场景和一个场景的详细示例。无论你是刚入门的新手，还是正在迁移旧项目的开发者，这篇指南都能帮你快速定位所需知识。

💡 提示：所有示例均假设已安装对应包（如 langchain-core、langchain-openai），并正确设置了 API 密钥。

一、核心抽象与运行时

这是 LangChain 生态的基石，所有组件都遵循 Runnable 接口，通过 LCEL 的 | 操作符无缝组合。

  • **Runnable** 接口
    功能：所有可组合组件的统一接口。
    用法：实现该接口的组件必须提供 invoke、stream、batch 等方法。
    适用场景：

    1. 任何需要被链式调用的自定义组件。

    2. 需要统一调用方式（同步/异步/批处理/流式）的模块。

举例：假设你写了一个自定义的文本清洗函数，想把它接入 LCEL 链。可以通过 RunnableLambda 包装，而 RunnableLambda 本身就是 Runnable 的实现，因此可以无缝与提示模板、模型组合。

  • **RunnableSequence**
    功能：顺序执行多个 Runnable。通常通过 | 隐式创建。
    用法：直接使用 LCEL 管道符即可。
    适用场景：

    1. 多步数据处理流水线（如清洗 → 翻译 → 总结）。

    2. 任何需要按固定顺序执行多个操作的任务。

举例：在问答系统中，先检索相关文档，再将文档和问题一起送给模型生成答案。这可以用 retriever | prompt | model 隐式构成序列。

  • **RunnableParallel**
    功能：并行执行多个 Runnable，返回包含所有结果的字典。
    用法：同时执行多个任务，提高效率。
    适用场景：

    1. 同时进行多路检索（如向量检索 + 关键词检索）。

    2. 对同一输入做多角度分析（如情感分析 + 关键词提取）。

举例：在客服系统中，用户问题同时送去检索常见问题库、查询订单状态、分析情绪，然后汇总结果。

chain = RunnableParallel(
    faq=faq_retriever,
    order=order_query,
    sentiment=sentiment_analyzer
)
result = chain.invoke("我的订单怎么还没到？")

  • **RunnableLambda**
    功能：将普通 Python 函数转换为 Runnable，嵌入 LCEL 链。
    用法：用于自定义数据处理。
    适用场景：

    1. 数据清洗、格式转换。

    2. 调用外部非 LangChain 库。

    3. 添加日志或调试信息。

举例：在生成回复前，需要将用户输入中的敏感信息（如手机号）脱敏。

def mask_phone(text: str) -> str:
    import re
    return re.sub(r'\d{11}', '***********', text)
clean = RunnableLambda(mask_phone)
chain = clean | prompt | model
  • **RunnableBranch**
    功能：根据条件动态选择要执行的 Runnable。
    用法：一组条件函数 + 默认分支。
    适用场景：

  1. 根据问题长度选择不同的处理流程（长文摘要 vs 短问答）。

  2. 根据用户身份（VIP/普通）调用不同的模型或策略。

举例：对于简短查询直接回答，对于长查询先做检索再回答。

branch = RunnableBranch(
    (lambda x: len(x) < 50, direct_chain),
    (lambda x: len(x) < 500, rag_chain),
    summary_chain  # 默认
)
  • **RunnablePassthrough**
    功能：传递输入数据不变，常与 RunnableParallel 配合保留原始数据。
    用法：{"key": RunnablePassthrough()} 表示将整个输入作为 key 的值。
    适用场景：

  1. 在并行分支中需要原始输入作为上下文。

  2. 添加新字段时不覆盖原有数据。

举例：RAG 链中既需要原始问题，又需要检索到的文档。

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
)
  • **.invoke()** 方法
    功能：同步调用 Runnable 的标准方法。
    用法：输入字典或单值，执行链。
    适用场景：

  1. 任何需要立即得到结果的调用（如 API 响应）。

  2. 脚本或命令行工具中的主流程。

举例：在 Web 后端中处理用户请求。

result = chain.invoke({"user_input": "今天天气怎么样？"})

  • **.ainvoke()** 方法
    功能：异步调用 Runnable。
    用法：适用于 FastAPI 等异步环境。
    适用场景：

    1. 高并发 Web 服务，避免阻塞事件循环。

    2. 需要同时处理多个请求的异步框架。

举例：在 FastAPI 路由中使用。

@app.post("/chat")
async def chat_endpoint(request: Request):
    data = await request.json()
    result = await chain.ainvoke(data)
    return {"answer": result}

  • **.stream()** 方法
    功能：流式返回输出（如逐字生成文本）。
    用法：迭代处理流式块。
    适用场景：

    1. 聊天机器人逐字显示回复。

    2. 长文本生成时提供实时反馈。

for chunk in chain.stream({"query": "讲个笑话"}):
    print(chunk, end="", flush=True)
  • **.batch()** 方法
    功能：批量处理一组输入，自动并行化。
    用法：传入输入列表。
    适用场景：

  1. 离线批量处理数据（如对数千条评论做情感分析）。

  2. 测试或评估时批量运行用例。

举例：批量翻译多个句子。

sentences = ["Hello", "How are you?", "Goodbye"]
translations = translate_chain.batch([{"text": s} for s in sentences])
  • **.with_retry()** 方法
    功能：为 Runnable 添加重试机制。
    用法：配置重试次数等。
    适用场景：

  1. 调用外部 API 可能临时失败（如网络抖动）。

  2. 重要任务需要保证成功率。

举例：调用 OpenAI 模型时偶尔遇到 5xx 错误，自动重试 3 次。

reliable_model = model.with_retry(stop_after_attempt=3)
result = reliable_model.invoke("讲个故事")
  • **.with_fallbacks()** 方法
    功能：定义备用 Runnable（主组件失败时自动切换）。
    用法：实现故障转移。
    适用场景：

  1. 主模型不可用时切换到备用模型（如从 GPT-4 降级到 GPT-3.5）。

  2. 多云部署，一个服务商故障时切换到另一个。

举例：OpenAI 服务异常时回退到本地模型。

fallback_model = ChatOllama(model="llama2")
resilient_chain = model.with_fallbacks([fallback_model])
  • **.configurable_fields()**
    功能：允许在运行时配置 Runnable 的特定字段（如 temperature）。
    用法：与 ConfigurableField 配合。
    适用场景：

  1. 同一个链在不同场景下需要不同参数（如创意写作 vs 事实问答）。

  2. 用户可在前端调整模型温度、top_p 等。

举例：让用户通过请求参数控制模型的随机性。

from langchain_core.runnables import ConfigurableField
model = ChatOpenAI().configurable_fields(
    temperature=ConfigurableField(id="temp", name="温度", description="控制随机性")
)
chain = prompt | model
# 调用时通过 config 指定
result = chain.invoke({"query": "写首诗"}, config={"configurable": {"temp": 0.8}})

  • **ConfigurableField**
    功能：标记一个字段为可配置。
    用法：定义字段 ID、名称、描述。
    适用场景：

    1. 与 configurable_fields 配合使用，定义可配置参数。

    2. 构建可动态调整的组件库。

举例：同上。

  • **RunnableConfig**
    功能：调用时传递的配置对象（回调、标签、可配置字段值等）。
    用法：通过 config 参数传入。
    适用场景：

    1. 为单次调用添加回调处理器。

    2. 传递可配置字段的值。

    3. 添加标签用于后续过滤追踪。

举例：调用时带上标签，方便在 LangSmith 中筛选。

chain.invoke({"text": "你好"}, config={"tags": ["user-facing", "chinese"]})

二、模型 I/O

管理与大语言模型的交互：提示构建、模型调用、输出解析。

  • **ChatOpenAI**（from langchain_openai）
    功能：调用 OpenAI 聊天模型。
    用法：需设置 OPENAI_API_KEY。
    适用场景：

    1. 需要强大通用能力的对话、生成、推理任务。

    2. 需要函数调用、结构化输出等高级功能。

举例：构建一个智能客服助手。

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
response = model.invoke([HumanMessage(content="我的订单号是12345，请问到哪了？")])
  • **ChatPromptTemplate**
    功能：构建聊天消息列表的提示模板。
    用法：from_messages 传入角色和内容。
    适用场景：

  1. 多轮对话场景（系统消息、历史、用户问题）。

  2. 需要给模型设定角色或行为。

举例：一个法律咨询助手，需要系统消息设定身份。

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名资深律师，回答要严谨且引用法律条文。"),
    ("user", "问题：{question}")
])
chain = prompt | model
  • **PromptTemplate**
    功能：构建单条字符串提示。
    用法：适用于非对话式生成。
    适用场景：

  1. 文本生成任务（摘要、翻译、扩写）。

  2. 简单的单轮问答。

举例：邮件自动回复生成。

prompt = PromptTemplate.from_template("根据以下邮件内容，生成一个礼貌的回复草稿：\n{email_content}")
chain = prompt | model | StrOutputParser()
  • **HumanMessagePromptTemplate**
    功能：专门创建用户角色的消息模板。
    用法：用于复杂提示构建。
    适用场景：

  1. 需要动态生成用户消息内容。

  2. 在构建消息列表时与系统消息区分。

举例：同上，但使用更细粒度的构建方式。

from langchain.prompts import HumanMessagePromptTemplate
human_template = HumanMessagePromptTemplate.from_template("{question}")
prompt = ChatPromptTemplate.from_messages([system_template, human_template])
  • **SystemMessagePromptTemplate**
    功能：专门创建系统角色的消息模板。
    用法：设定 AI 行为。
    适用场景：

  1. 需要根据上下文动态调整系统消息。

  2. 多角色场景。

举例：根据用户选择的角色动态设定系统消息。

system_template = SystemMessagePromptTemplate.from_template("你是一位{role}专家，回答要专业。")
  • **MessagesPlaceholder**
    功能：在提示模板中为可变长度消息列表预留位置。
    用法：常用于插入对话历史或 Agent 中间步骤。
    适用场景：

  1. 多轮对话中插入历史消息。

  2. Agent 的 agent_scratchpad 占位。

举例：带历史记录的聊天机器人。

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手。"),
    MessagesPlaceholder("chat_history"),
    ("user", "{input}")
])
  • **.format()** 方法（PromptTemplate）
    功能：同步格式化提示模板。
    用法：填充变量，得到最终提示字符串或消息列表。
    适用场景：

  1. 需要预览生成的提示内容。

  2. 在不调用模型的情况下获取提示。

举例：调试时检查提示是否构造正确。

filled = prompt.format(question="今天天气如何？")
print(filled)
  • **.format_prompt()** 方法
    功能：格式化后返回 PromptValue 对象。
    用法：可方便转换为字符串或消息列表。
    适用场景：

  1. 在链中传递格式化的提示值。

  2. 需要同时支持字符串和消息格式的场景。

举例：与多种模型类型兼容。

prompt_val = prompt.format_prompt(question="你好")
# 可以转为字符串或消息列表
messages = prompt_val.to_messages()
  • **StrOutputParser**
    功能：最简单的输出解析器，提取 AIMessage 内容为字符串。
    用法：链末尾常用。
    适用场景：

  1. 绝大多数需要纯文本输出的场景。

  2. 与流式输出配合，逐字返回字符串。

举例：直接返回模型生成的文本。

chain = model | StrOutputParser()
  • **JsonOutputParser**
    功能：解析模型输出为 JSON 对象。
    用法：结合提示要求模型输出 JSON。
    适用场景：

  1. 需要结构化数据（如提取字段、生成配置）。

  2. 与 API 交互时希望直接获得 JSON。

举例：从简历文本中提取姓名、电话、邮箱。

parser = JsonOutputParser()
prompt = PromptTemplate.from_template(
    "从以下文本提取信息，输出JSON格式包含name, phone, email字段。\n文本：{text}\n{format_instructions}",
    partial_variables={"format_instructions": parser.get_format_instructions()}
)
chain = prompt | model | parser
  • **PydanticOutputParser**
    功能：根据 Pydantic 模型解析输出，确保强类型。
    用法：定义数据类，传入 parser。
    适用场景：

  1. 需要严格类型检查和自动验证。

  2. 下游系统要求特定数据结构。

举例：解析用户意图和参数。

from pydantic import BaseModel, Field
class Intent(BaseModel):
    action: str = Field(description="用户意图，如'查询天气'、'设置闹钟'")
    params: dict = Field(description="参数")
parser = PydanticOutputParser(pydantic_object=Intent)
prompt = PromptTemplate(template="分析用户输入：{input}\n{format_instructions}", partial_variables={"format_instructions": parser.get_format_instructions()})
chain = prompt | model | parser
result = chain.invoke({"input": "明天早上八点叫我起床"})
print(result.action)  # 输出: 设置闹钟
  • **CommaSeparatedListOutputParser**
    功能：将逗号分隔的文本解析为列表。
    用法：适合生成列表项。
    适用场景：

  1. 让模型列举要点。

  2. 生成关键词列表。

举例：生成一篇关于AI的文章的关键词。

parser = CommaSeparatedListOutputParser()
prompt = PromptTemplate(template="列举人工智能的五个应用领域，用逗号分隔。\n{format_instructions}", partial_variables={"format_instructions": parser.get_format_instructions()})
chain = prompt | model | parser
keywords = chain.invoke({})
print(keywords)  # ['自然语言处理', '计算机视觉', ...]

  • **get_format_instructions()**
    功能：输出解析器的方法，返回指导模型格式化的指令文本。
    用法：插入提示中，告知模型输出格式。
    适用场景：

    1. 动态生成格式说明，减少硬编码。

    2. 与各种解析器配合使用。
      举例：见上。

  • **init_chat_model()**（from langchain.chat_models）
    功能：根据模型名字符串自动初始化对应模型类。
    用法：简化代码，无需关心具体是 OpenAI 还是 Anthropic。
    适用场景：

    1. 需要根据配置动态切换模型提供商。

    2. 编写与模型无关的通用代码。

举例：从环境变量读取模型名称，自动创建实例。

model_name = os.getenv("MODEL", "gpt-4o")
model = init_chat_model(model_name, temperature=0.5)
  • **BaseMessage**
    功能：所有消息的基类。
    用法：常见子类 HumanMessage、AIMessage、SystemMessage、ToolMessage。
    适用场景：

  1. 手动构造消息列表。

  2. 处理模型返回的消息。

举例：构建包含系统消息和用户消息的列表。

from langchain_core.messages import SystemMessage, HumanMessage
messages = [SystemMessage(content="你是一个物理老师"), HumanMessage(content="什么是熵？")]
  • **.content** 属性（BaseMessage）
    功能：消息的文本内容。
    用法：从模型返回的 AIMessage 中提取内容。
    适用场景：

  1. 获取模型生成的文本。

  2. 处理历史消息。

举例：

response = model.invoke(messages)
answer = response.content
  • **.tool_calls** 属性（AIMessage）
    功能：当模型调用工具时，包含结构化工具调用信息。
    用法：Agent 执行器依赖此信息执行工具。
    适用场景：

  1. 构建自定义 Agent 时解析工具调用。

  2. 调试工具调用是否按预期发生。

举例：

if response.tool_calls:
    for tool_call in response.tool_calls:
        print(f"调用工具 {tool_call['name']}，参数 {tool_call['args']}")

三、检索增强生成（RAG）

连接外部数据，让模型“开卷考试”。

  • **Document**（from langchain_core.documents）
    功能：表示一个文档，包含 page_content 和 metadata。
    用法：数据在加载、转换、检索、生成全流程中的标准格式。
    适用场景：

    1. 存储从文件、网页加载的文本块。

    2. 携带来源、页码等元信息。

举例：从 PDF 加载后，每个页面作为一个 Document。

doc = Document(page_content="这是内容", metadata={"source": "file.pdf", "page": 1})
  • **BaseLoader**
    功能：所有文档加载器的基类。
    用法：从不同来源加载文档。
    适用场景：

  1. 需要从特定源加载数据（文件、数据库、网络）。

  2. 自定义数据源。

举例：实现一个从数据库加载文档的自定义加载器。

class MyDBLoader(BaseLoader):
    def load(self):
        # 从数据库读取并返回 Document 列表
        ...
  • **TextLoader**
    功能：从文本文件加载文档。
    用法：整个文件作为一个 Document。
    适用场景：

  1. 处理 .txt 文件。

  2. 快速加载小文本。

举例：

loader = TextLoader("notes.txt")
docs = loader.load()
  • **PyPDFLoader**
    功能：加载 PDF 文件，每页作为一个 Document。
    用法：处理 PDF 文档。
    适用场景：

  1. 处理 PDF 报告、论文。

  2. 需要对 PDF 内容进行问答或摘要。

举例：

loader = PyPDFLoader("annual_report.pdf")
pages = loader.load()  # 每页一个 Document
  • **RecursiveCharacterTextSplitter**
    功能：最常用的文本分割器，递归尝试不同分隔符。
    用法：设置 chunk_size 和 chunk_overlap。
    适用场景：

  1. 将长文档分割成适合嵌入和检索的块。

  2. 任何需要分块的文本处理。

举例：将一本书的章节分割成约 500 字的段落。

splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(pages)
  • **CharacterTextSplitter**
    功能：基于单个字符分隔符的简单分割器。
    用法：按指定分隔符分割。
    适用场景：

  1. 文本已经按固定分隔符组织（如每行一条记录）。

  2. 对分割粒度要求不高时。

举例：按换行符分割日志文件。

splitter = CharacterTextSplitter(separator="\n", chunk_size=1000)
  • **OpenAIEmbeddings**
    功能：调用 OpenAI 嵌入模型，将文本转为向量。
    用法：需设置 API 密钥。
    适用场景：

  1. 需要高质量语义向量的 RAG 应用。

  2. 与向量数据库配合。

举例：

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector = embeddings.embed_query("查询文本")
  • **FAISS**（from langchain_community.vectorstores）
    功能：本地向量数据库封装。
    用法：from_documents 创建索引，as_retriever 转为检索器。
    适用场景：

  1. 轻量级本地 RAG，无需额外服务。

  2. 原型验证或小规模数据。

举例：

vectorstore = FAISS.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

  • **.from_documents()**（VectorStore）
    功能：从 Document 列表创建向量存储。
    用法：自动调用嵌入模型并存储。
    适用场景：

    1. 初始构建索引。

    2. 增量添加文档（部分向量库支持）。

举例：同上。

  • **.as_retriever()**（VectorStore）
    功能：将向量存储对象转换为检索器（Retriever）。
    用法：获得统一的 .invoke() 接口。
    适用场景：

    1. 将向量存储作为检索器接入 LCEL 链。

    2. 配置检索参数（如返回数量）。

举例：

retriever = db.as_retriever(search_kwargs={"k": 5})
  • **.similarity_search()**（VectorStore）
    功能：执行相似性搜索。
    用法：返回最相似文档列表。
    适用场景：

  1. 直接获取相似文档而不经过链。

  2. 调试检索效果。

举例：

docs = db.similarity_search("机器学习", k=3)
  • **.similarity_search_with_relevance_scores()**
    功能：相似性搜索并返回文档及分数。
    用法：用于评估或设置阈值。
    适用场景：

  1. 需要过滤低分文档。

  2. 评估检索质量。

举例：

docs_and_scores = db.similarity_search_with_relevance_scores("query")
filtered = [(doc, score) for doc, score in docs_and_scores if score > 0.7]
  • **BaseRetriever**
    功能：所有检索器的基类。
    用法：必须实现 .invoke() 方法。
    适用场景：

  1. 自定义检索逻辑（如从数据库、API 检索）。

  2. 包装现有检索系统。

举例：

class MyRetriever(BaseRetriever):
    def _get_relevant_documents(self, query: str, *, run_manager):
        # 实现自定义检索
        return [Document(...)]
  • **.as_retriever()** 参数 **search_type**
    功能：指定检索类型，如 "similarity"（相似度）、"mmr"（最大边际相关性，增加多样性）。
    用法：search_type="mmr"。
    适用场景：

  1. 需要结果多样性时（如推荐系统）。

  2. 避免返回过于相似的文档。

举例：

retriever = db.as_retriever(search_type="mmr", search_kwargs={"k": 5, "fetch_k": 20})

  • **.as_retriever()** 参数 **search_kwargs**
    功能：提供额外检索参数，如 {"k": 5}、{"fetch_k": 20}（用于 MMR）。
    用法：search_kwargs={"k": 10}。
    适用场景：

    1. 调整返回数量。

    2. 传递特定检索算法的参数。
      举例：同上。

  • **EnsembleRetriever**
    功能：集成多个检索器（如 BM25 + 向量检索），提高召回率。
    用法：传入检索器列表和权重。
    适用场景：

    1. 需要结合关键词和语义检索优势。

    2. 对召回率要求高的场景。
      举例：混合 BM25 和向量检索。

from langchain.retrievers import BM25Retriever, EnsembleRetriever
bm25 = BM25Retriever.from_texts(["text1", "text2"])
vector = db.as_retriever()
ensemble = EnsembleRetriever(retrievers=[bm25, vector], weights=[0.3, 0.7])
  • **MultiQueryRetriever**
    功能：使用 LLM 生成多个视角的查询，分别检索后合并。
    用法：解决查询表述不佳的问题。
    适用场景：

  1. 用户查询可能不准确或不完整。

  2. 需要从多个角度理解问题。

举例：用户问“如何学习编程”，生成“编程入门方法”、“初学者编程资源”、“编程学习路径”等查询。

retriever = MultiQueryRetriever.from_llm(retriever=base_retriever, llm=llm)
docs = retriever.invoke("如何学习编程")
  • **ContextualCompressionRetriever**
    功能：对检索出的文档进行压缩（提取关键部分），减少上下文长度。
    用法：配合 LLMChainExtractor 或 EmbeddingsFilter。
    适用场景：

  1. 模型上下文窗口有限。

  2. 文档噪音大，只需关键信息。

举例：用 LLM 提取每个文档中与问题最相关的句子。

from langchain.retrievers.document_compressors import LLMChainExtractor
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(base_compressor=compressor, base_retriever=retriever)
  • **create_retrieval_chain**
    功能：构建标准 RAG 链（接收用户输入，检索，组合文档与输入后交给 LLM）。
    用法：与 create_stuff_documents_chain 配合。
    适用场景：

  1. 快速搭建问答系统。

  2. 需要将文档合并到提示中的标准流程。

举例：

from langchain.chains import create_retrieval_chain, create_stuff_documents_chain
combine_docs_chain = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(retriever, combine_docs_chain)
result = rag_chain.invoke({"input": "什么是LangChain？"})
  • **create_history_aware_retriever**
    功能：创建能结合对话历史进行检索的检索器（自动生成独立查询）。
    用法：用于多轮对话 RAG。
    适用场景：

  1. 多轮对话中需要理解指代（如“它”指什么）。

  2. 需要基于历史生成新的检索查询。

举例：

from langchain.chains import create_history_aware_retriever
prompt = ChatPromptTemplate.from_messages([
    MessagesPlaceholder("chat_history"),
    ("user", "{input}"),
    ("user", "根据对话，生成一个用于检索的独立问题。")
])
history_aware_retriever = create_history_aware_retriever(llm, retriever, prompt)

四、链与执行逻辑

LCEL 的声明式组合与高级控制流。

  • **Runnable** 管道符 **|**
    功能：LCEL 核心操作符，连接两个 Runnable。
    用法：前一个的输出作为后一个的输入。
    适用场景：

    1. 任何需要将多个处理步骤串联的场景。

    2. 构建模块化、可读性强的链。
      举例：chain = retriever | prompt | model | parser。

  • 创建基础链
    功能：提示模板、模型、输出解析器连接。
    示例：chain = prompt_template | model | StrOutputParser()。
    适用场景：任何单轮生成任务，如翻译、摘要。

  • 创建 RAG 链
    功能：检索器、提示、模型、解析器连接。
    示例：chain = {"context": retriever, "question": RunnablePassthrough()} | prompt | model | parser。
    适用场景：问答系统、文档聊天。

  • **RunnableParallel** 用于分支
    功能：同一输入并行执行多个处理。
    示例：analysis_chain = RunnableParallel(summary=sum_chain, sentiment=sent_chain)。
    适用场景：对同一数据做多维度分析，如评论的情感、关键词、摘要。

  • **.map()** 方法（Runnable）
    功能：将 Runnable 应用于输入列表的每个元素。
    用法：批量处理列表。
    适用场景：

    1. 需要对列表中的每个元素应用同一处理（如对多个文档分别摘要）。

    2. 并行化处理独立任务。

举例：对一批新闻标题进行情感分析。

sentiment_chain = prompt | model | parser
batch_chain = sentiment_chain.map()
results = batch_chain.batch([{"title": t} for t in titles])
  • **itemgetter**（from operator）
    功能：从输入字典中提取特定键的值，比 lambda 更高效。
    用法：与 RunnableParallel 配合。
    适用场景：

  1. 需要从字典中提取字段作为后续步骤的输入。

  2. 避免编写 lambda 函数，提高性能。

举例：

from operator import itemgetter
chain = RunnableParallel(context=itemgetter("doc"), question=itemgetter("q")) | prompt
  • **Runnable.assign()**
    功能：在现有字典上添加新字段（或覆盖）。
    用法：不丢失原有数据，丰富数据流。
    适用场景：

  1. 在保留原始输入的同时添加处理结果。

  2. 逐步构建更丰富的上下文。

举例：先检索，再在原有输入上添加上下文。

chain = RunnablePassthrough.assign(context=retriever) | prompt
  • **Runnable.bind()**
    功能：绑定额外参数（如工具定义、response_format）。
    用法：常用于模型工具调用。
    适用场景：

  1. 在调用模型前绑定工具列表。

  2. 绑定固定的生成参数（如 stop 序列）。

举例：

tools = [...]  # 工具列表
model_with_tools = model.bind(tools=tools)
chain = prompt | model_with_tools
  • **Runnable.with_types()**
    功能：指定 Runnable 的输入和输出类型（用于类型检查、OpenAPI 文档）。
    用法：传入 Pydantic 模型。
    适用场景：

  1. 为链定义清晰的接口，便于文档生成。

  2. 类型验证，防止错误数据流入。

举例：

from pydantic import BaseModel
class InputSchema(BaseModel):
    question: str
class OutputSchema(BaseModel):
    answer: str
typed_chain = chain.with_types(input_type=InputSchema, output_type=OutputSchema)
  • **RunnablePick**（或 .pick()）
    功能：从输入字典中选取一个或多个键，只传递这些键。
    用法：精简数据。
    适用场景：

  1. 后续步骤只需要部分字段，避免传递无关数据。

  2. 从多字段中提取所需字段。

举例：

chain = RunnablePick("user_id", "query") | next_step
  • **RouterRunnable**
    功能：根据输入中的键值将请求路由到不同 Runnable。
    用法：键值映射路由。
    适用场景：

  1. 根据任务类型调用不同的处理链（如翻译、摘要、问答）。

  2. 多租户系统中根据不同租户配置调用不同逻辑。

举例：

router = RouterRunnable({
    "translate": translate_chain,
    "summarize": summarize_chain,
    "qa": qa_chain
})
result = router.invoke({"key": "translate", "input": "Hello world", "target_lang": "zh"})

五、记忆（Memory）

为有状态对话应用提供历史记录管理。

  • **BaseMemory**
    功能：所有记忆类的基类。
    用法：实现 load_memory_variables 和 save_context。
    适用场景：

    1. 自定义记忆存储（如用 Redis 存储）。

    2. 需要特殊记忆逻辑的应用。

举例：实现一个将记忆存到文件的简单记忆。

class FileMemory(BaseMemory):
    def __init__(self, filepath):
        self.filepath = filepath
    def load_memory_variables(self, inputs):
        with open(self.filepath, 'r') as f:
            return {"history": f.read()}
    def save_context(self, inputs, outputs):
        with open(self.filepath, 'a') as f:
            f.write(f"User: {inputs['input']}\nAI: {outputs['output']}\n")
  • **ConversationBufferMemory**
    功能：最简单记忆，直接缓存所有历史消息。
    用法：适合简短对话，注意 token 增长。
    适用场景：

  1. 短期对话（如单次会话）。

  2. 测试或原型开发。

举例：

memory = ConversationBufferMemory(return_messages=True)
memory.save_context({"input": "你好"}, {"output": "你好！"})
memory.save_context({"input": "你叫什么？"}, {"output": "我是AI助手。"})
history = memory.load_memory_variables({})["history"]
  • **ConversationBufferWindowMemory**
    功能：只保留最近 k 轮对话。
    用法：限制窗口大小。
    适用场景：

  1. 需要控制 token 消耗，且最近对话最重要。

  2. 长对话中只关注近期上下文。

举例：只保留最近 3 轮。

memory = ConversationBufferWindowMemory(k=3, return_messages=True)
  • **ConversationSummaryMemory**
    功能：用 LLM 定期总结历史，存储总结。
    用法：减少 token，但增加一次模型调用。
    适用场景：

  1. 长对话，需要保留长期记忆但不想浪费 token。

  2. 对记忆准确性要求不高，可接受摘要。

举例：

memory = ConversationSummaryMemory(llm=llm, return_messages=True)
memory.save_context({"input": "我喜欢吃苹果"}, {"output": "好的，记下了"})
summary = memory.load_memory_variables({})["history"]  # 可能是总结
  • **ConversationSummaryBufferMemory**
    功能：结合 Buffer 和 Summary：超阈值时触发总结。
    用法：设置 max_token_limit。
    适用场景：

  1. 需要平衡近期细节和长期摘要。

  2. 对话长度变化大，希望自适应压缩。

举例：

memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=2000, return_messages=True)
  • **VectorStoreRetrieverMemory**
    功能：使用向量存储检索与当前对话最相关的历史片段。
    用法：适合长期、大量记忆。
    适用场景：

  1. 需要长期记忆，但无法保留全部历史。

  2. 记忆需要按相关性检索的场景。

举例：

from langchain.memory import VectorStoreRetrieverMemory
memory = VectorStoreRetrieverMemory(retriever=vectorstore.as_retriever())
memory.save_context({"input": "我生日是5月1日"}, {"output": "记下了"})
# 之后问“我的生日是什么时候？”时，会检索到相关记忆
  • **EntityMemory**
    功能：从对话中提取并存储关于特定实体（人物、地点）的信息。
    用法：记住用户偏好或重要实体。
    适用场景：

  1. 需要记住用户个人信息（如名字、喜好）。

  2. 知识密集型对话，需维护实体知识库。

举例：

memory = EntityMemory(llm=llm)
memory.save_context({"input": "我叫张三"}, {"output": "你好张三"})
# 后续问“我叫什么？”时会从实体记忆中提取
  • **load_memory_variables()**
    功能：从记忆中加载历史变量。
    用法：返回字典，键常为 history。
    适用场景：

  1. 在链中加载记忆并注入提示。

  2. 手动获取记忆内容用于显示。

举例：

vars = memory.load_memory_variables({"input": "当前问题"})
history = vars["history"]
  • **save_context()**
    功能：将一轮对话的输入和输出保存到记忆。
    用法：每次交互后调用。
    适用场景：

  1. 对话结束后保存。

  2. 在链中手动调用。

举例：

memory.save_context({"input": "用户输入"}, {"output": "模型输出"})

  • **clear()**
    功能：清空记忆。
    用法：重置对话状态。
    适用场景：

    1. 开始新对话时清空旧记忆。

    2. 测试时重置状态。

举例：memory.clear()

  • 在 LCEL 链中使用记忆
    功能：通过 RunnableLambda 手动加载和保存记忆。
    示例：
def load_memory(input_dict):
    history = memory.load_memory_variables({})["chat_history"]
    return {**input_dict, "history": history}
def save_memory(input_dict, output):
    memory.save_context({"input": input_dict["input"]}, {"output": output})
    return output
chain = RunnableLambda(load_memory) | prompt | model | RunnableLambda(save_memory)
  • 记忆与 LCEL 集成示例

功能：展示如何在链开头加载记忆，末尾保存。

示例：同上。

六、工具与智能体（Tools & Agents）

让 LLM 能够调用外部工具，执行复杂任务。

  • **@tool** 装饰器
    功能：将普通函数转换为 Tool 对象。
    用法：文档字符串成为工具描述。
    适用场景：

    1. 快速将现有函数包装成工具。

    2. 开发简单工具时避免手动创建 Tool 类。

举例：

@tool
def get_weather(city: str) -> str:
    """返回指定城市的天气（模拟）"""
    return f"{city} 今天晴天，20度"
  • **Tool** 类
    功能：表示一个工具，包含 name、description、func。
    用法：可手动创建。
    适用场景：

  1. 需要更精细控制工具属性时。

  2. 包装已有函数，但需要自定义名称和描述。

举例：

tool = Tool(name="Calculator", func=eval, description="计算数学表达式")
  • **StructuredTool** 类
    功能：更强大的工具类，支持复杂输入参数模式。
    用法：适用于多参数、有类型要求的函数。
    适用场景：

  1. 工具需要多个参数，且需要类型验证。

  2. 与 Pydantic 模型结合，生成更好的工具描述。

举例：

from pydantic import BaseModel, Field
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    num_results: int = Field(description="返回结果数量", default=5)
def search(query: str, num_results: int = 5):
    ...
tool = StructuredTool.from_function(func=search, args_schema=SearchInput)
  • **load_tools()**
    功能：加载 LangChain 预定义的工具集。
    用法：传入工具名称列表，如 ["llm-math", "serpapi"]。
    适用场景：

  1. 快速集成常见工具（如搜索引擎、计算器）。

  2. 原型开发时无需自己实现。

举例：

tools = load_tools(["llm-math", "wikipedia"], llm=llm)
  • **Toolkit**
    功能：一组为特定任务设计的工具集合（如 SQLDatabaseToolkit）。
    用法：直接使用预定义 toolkit。
    适用场景：

  1. 与特定数据源交互（数据库、Gmail、Slack）。

  2. 需要多个相关工具的场景。

举例：

from langchain_community.agent_toolkits import SQLDatabaseToolkit
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
tools = toolkit.get_tools()
  • **create_react_agent()**
    功能：创建 ReAct 范式智能体（Reason + Act）。
    用法：交替思考与行动。
    适用场景：

  1. 需要多步推理的任务（如复杂问题解决）。

  2. 工具调用频繁且需要推理中间步骤。

举例：

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
result = executor.invoke({"input": "我需要预订一张明天去北京的火车票，并查询当地天气"})

  • **create_openai_functions_agent()**
    功能：针对 OpenAI 函数调用能力优化的智能体。
    用法：利用原生函数调用，更稳定。
    适用场景：

    1. 使用 OpenAI 模型且需要工具调用。

    2. 希望减少提示工程，利用模型原生能力。
      举例：同上，但使用函数调用版本。

  • **create_tool_calling_agent()**
    功能：通用的工具调用智能体创建函数（推荐）。
    用法：适用于支持工具调用的模型（OpenAI、Anthropic 等）。
    适用场景：

    1. 需要兼容多种模型提供商的工具调用。

    2. 推荐在新项目中使用。
      举例：

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
  • **AgentExecutor**
    功能：智能体的执行器，循环驱动思考、调用工具、处理结果。
    用法：传入 agent 和 tools。
    适用场景：

  1. 执行任何智能体。

  2. 控制循环参数（迭代次数、错误处理）。

举例：

executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=10)

  • **AgentExecutor** 参数 **handle_parsing_errors**
    功能：处理输出解析错误的策略（True 将错误信息返回给 agent 修正）。
    用法：AgentExecutor(..., handle_parsing_errors=True)。
    适用场景：

    1. 模型偶尔输出格式错误，希望它自行修正。

    2. 提高鲁棒性。

  • **AgentExecutor** 参数 **max_iterations**
    功能：限制最大循环次数，防止无限循环。
    用法：AgentExecutor(..., max_iterations=5)。
    适用场景：

    1. 避免智能体陷入死循环或过长思考。

    2. 控制成本和时间。

  • **AgentExecutor** 参数 **early_stopping_method**
    功能：达到最大次数或无法继续时如何生成最终回复（"generate" 或 "error"）。
    用法：AgentExecutor(..., early_stopping_method="generate")。
    适用场景：

    1. 希望即使没完成也给出一个答案。

    2. 或直接报错。

  • **agent_scratchpad**
    功能：提示模板中的占位符，存储 Agent 中间思考过程和工具调用结果。
    用法：MessagesPlaceholder("agent_scratchpad")。
    适用场景：

    1. 构建 Agent 提示时必须包含此占位。

    2. 用于传递中间步骤给模型。

  • **create_agent()**（高层 API）
    功能：简化 Agent 创建，只需模型和工具。
    用法：JetBrains 博客提及的简化方式。
    适用场景：

    1. 快速搭建原型。

    2. 内部工具，不关心细节。
      举例：agent = create_agent("gpt-5", tools=tools)（假设存在）。

  • **AgentType**（旧版）
    功能：在 initialize_agent 中指定智能体类型（如 ZERO_SHOT_REACT_DESCRIPTION）。
    用法：新代码推荐使用 create_xxx_agent。
    适用场景：阅读旧代码时需要理解。

  • **ModelFallbackMiddleware**
    功能：中间件，主模型失败时切换到备用模型。
    用法：通过 middleware 参数配置。
    适用场景：

    1. 提高服务可用性。

    2. 多模型备份。
      举例：

from langchain.agents import create_react_agent, AgentExecutor
from langchain.agents.middleware import ModelFallbackMiddleware
middleware = [ModelFallbackMiddleware(fallback_models=[backup_llm])]
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, middleware=middleware)

七、回调与观测（Callbacks & Observability）

监控、日志、追踪 LLM 应用的运行。

  • **BaseCallbackHandler**
    功能：所有回调处理器的基类。
    用法：继承并实现 on_llm_start、on_chain_end 等方法。
    适用场景：

    1. 自定义日志记录。

    2. 将追踪数据发送到自定义后端。

举例：

class MyCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"LLM 开始，提示：{prompts}")
    def on_chain_end(self, outputs, **kwargs):
        print(f"链结束，输出：{outputs}")
  • **StdOutCallbackHandler**
    功能：内置的将事件输出到标准输出的处理器。
    用法：开发和调试用。
    适用场景：

  1. 本地调试，查看执行过程。

  2. 快速了解链内部发生了什么。

举例：

from langchain_core.callbacks import StdOutCallbackHandler
chain.invoke(input, callbacks=[StdOutCallbackHandler()])

  • **callbacks** 参数
    功能：在 invoke、stream 等方法中传递回调处理器。
    用法：为单次调用指定回调。
    适用场景：

    1. 需要为特定调用添加特殊观测。

    2. 避免全局配置。

  • **CallbackManager**
    功能：管理和协调多个回调处理器的管理器。
    用法：通常不需要直接使用。
    适用场景：

    1. 高级自定义，需要控制回调的调度。

  • **run_id**
    功能：每次调用产生的唯一标识符。
    用法：在回调函数中用于关联所有事件。
    适用场景：

    1. 将多个事件归因到同一调用。

    2. 调试时追踪完整流程。

  • **.with_retry()** 的回调
    功能：重试事件也会触发回调。
    用法：监听 on_retry 事件。
    适用场景：

    1. 监控重试发生频率和原因。

  • LangSmith 集成
    功能：LangSmith 平台自动追踪和调试。
    用法：设置环境变量 LANGCHAIN_TRACING_V2=true。
    适用场景：

    1. 生产环境监控、调试。

    2. 团队协作，共享追踪数据。

八、数据预处理与后处理

文档转换、嵌入、输出修正等。

  • **BaseDocumentTransformer**
    功能：所有文档转换器（如文本分割器）的基类。
    用法：实现 transform_documents 方法。
    适用场景：

    1. 自定义文档转换逻辑（如去除页眉页脚）。

举例：实现一个去除空行的转换器。

class RemoveEmptyLines(BaseDocumentTransformer):
    def transform_documents(self, documents, **kwargs):
        for doc in documents:
            lines = [line for line in doc.page_content.split('\n') if line.strip()]
            doc.page_content = '\n'.join(lines)
        return documents
  • **HuggingFaceEmbeddings**
    功能：使用 Hugging Face 上的嵌入模型（本地运行）。
    用法：指定模型名称。
    适用场景：

  1. 需要本地部署，无需 API 调用。

  2. 使用开源嵌入模型。

举例：

embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-en-v1.5")
  • **OutputFixingParser**
    功能：第一次解析失败时，将错误和错误输出发给 LLM 修正后再试。
    用法：包装一个 parser。
    适用场景：

  1. 模型偶尔输出格式错误，希望自动修正。

  2. 对输出格式要求严格但希望容错。

举例：

fixing_parser = OutputFixingParser.from_llm(parser=parser, llm=llm)
chain = prompt | model | fixing_parser

  • **RetryOutputParser**
    功能：类似 OutputFixingParser，但连同原始提示一起重试，修正能力更强。
    用法：成本更高。
    适用场景：

    1. 输出必须完全符合格式，且第一次失败后需要重新生成。

    2. 对准确性要求极高。

九、LangGraph 生态进阶

LangGraph 是 LangChain 官方推出的用于构建有状态、多智能体应用的框架，以下组件在处理复杂流程时不可或缺。

  • **Command**（from langgraph）
    功能：从 LangGraph 节点内部发出指令，控制图的下一步行为（跳转、更新状态）。
    用法：实现复杂控制流（循环、条件跳转）。
    适用场景：

    1. 多智能体协作中动态决定下一步谁执行。

    2. 基于状态的条件路由。

举例：在客服智能体中，根据用户意图跳转到不同子图。

def router_node(state):
    intent = state["intent"]
    if intent == "退货":
        return Command(goto="return_process")
    elif intent == "咨询":
        return Command(goto="qa_process")
    else:
        return Command(goto="human_handoff")
  • **interrupt**（from langgraph.types）
    功能：在节点中暂停图执行，等待外部输入（人机交互）。
    用法：实现人工审批、信息补全。
    适用场景：

  1. 高风险操作需要人工审批。

  2. 模型需要用户提供额外信息。

举例：在转账智能体中，暂停等待用户确认。

from langgraph.types import interrupt
def transfer_node(state):
    amount = state["amount"]
    confirm = interrupt(f"请确认转账 {amount} 元？(y/n)")
    if confirm.lower() == 'y':
        return {"status": "transferred"}
    else:
        return {"status": "cancelled"}
  • **RetryPolicy**（from langgraph.types）
    功能：为 LangGraph 节点配置重试策略。
    用法：精细控制重试次数、退避时间、异常类型。
    适用场景：

  1. 节点可能暂时失败（如调用外部 API）。

  2. 需要自定义重试逻辑。

举例：

from langgraph.types import RetryPolicy
node = my_node.with_retry(policy=RetryPolicy(max_attempts=3, initial_interval=1.0))
  • **StateSnapshot**（from langgraph）
    功能：代表 LangGraph 在某个步骤开始时的状态快照。
    用法：用于调试、时间旅行。
    适用场景：

  1. 调试时查看中间状态。

  2. 实现“回放”功能。

举例：

config = {"configurable": {"thread_id": "1"}}
state = graph.get_state(config)
print(state.values)  # 当前状态

十、经典遗留链（Legacy Chains）

⚠️ 注意：以下组件来自 LangChain v0.1.x 时代，官方已不推荐在新项目中使用。它们被 LCEL 和新的创建函数取代。但你在维护老项目或参考旧教程时可能会遇到，特此列出以供理解。

  • **LLMChain**
    功能：旧版中最基础的链，组合提示模板和 LLM。
    替代：prompt | model。
    适用场景：维护旧代码。
    举例：
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(input="...")

  • **SimpleSequentialChain**
    功能：顺序执行多个链，前一个输出作为后一个输入。
    替代：chain1 | chain2。
    适用场景：旧项目迁移。

  • **ConversationalRetrievalChain**
    功能：对话式文档问答链，处理聊天历史。
    替代：create_history_aware_retriever + create_retrieval_chain。
    适用场景：旧项目中的对话式 RAG。

  • **load_summarize_chain**
    功能：专门用于文本摘要的链，支持 stuff、map_reduce、refine 模式。
    替代：LCEL 组合或使用 MapReduceDocumentsChain 等。
    适用场景：旧项目中的摘要功能。

  • **RetrievalQA**
    功能：简单的 RAG 链，不支持多轮对话历史。
    替代：create_retrieval_chain。
    适用场景：旧项目中的简单问答。

  • **APIChain**
    功能：允许 LLM 调用外部 API。
    替代：Agent 工具调用更灵活。
    适用场景：旧项目中的 API 调用。

🎯 总结与展望

至此，我们已全面梳理了 LangChain 生态中 122 个核心组件。从基础的 Runnable 接口，到 RAG 的完整流程，再到智能体和 LangGraph 的进阶玩法，相信你已经对 LangChain 的现在和未来有了清晰的认识。

核心要点回顾：

  • LCEL 是构建链的唯一推荐方式，所有组件都遵循 Runnable 协议。

  • **langchain-core** 定义了核心抽象，其他包提供具体实现。

  • 记忆、工具、回调等模块通过 LCEL 无缝集成。

  • LangGraph 正在成为复杂 Agent 应用的新标准。

  • 旧式 Chain 已逐步淘汰，新项目请拥抱 LCEL。

下一步建议：

  • 动手实践：从最简单的 prompt | model | parser 开始，逐步加入检索、记忆、工具。

  • 熟悉 LangSmith 的追踪功能，提升调试效率。

  • 探索 LangGraph 官方文档，学习构建多智能体系统。

# python # pycharm # 人工智能 # 自然语言处理 # 语言模型
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

从“看天吃饭”到“按图赚钱”：2026风光电站为什么必须重做气象服务体系

avatar AtomGit开源社区
cover

【Linux】Shell 脚本解析日志文件提取关键信息

avatar AtomGit开源社区
cover

零基础极速上手：用AI建站工具10分钟创建一个专业网站

avatar AtomGit开源社区