LangChain入门- 链 (Chains) 完全指南

标签: LangChain, Chains, 链系统, LCEL, 工作流, AI开发

深入讲解 LangChain 链系统，包括 LLMChain、SequentialChain、RouterChain、LCEL 组合方式、记忆集成和输出解析

链 (Chains)

---

用途

将多个组件（提示、模型、工具等）组合成端到端的工作流。链是 LangChain 的核心概念，用于构建复杂的应用逻辑。

位置

libs/langchain/langchain/chains/

知识图谱上下文

核心类/模块

基础链类

• Chain: 所有链的基类（继承自 RunnableSerializable）
• LLMChain: 最简单的链，提示 + LLM
• SequentialChain: 按顺序执行多个链
• TransformChain: 自定义数据转换
• RouterChain: 根据输入路由到不同子链

高级链

• RetrievalQA: 带检索的问答
• ConversationalRetrievalChain: 对话式检索问答
• SummarizationChain: 文档摘要
• RefineChain: 迭代优化输出
• ConstitutionalChain: 带约束的链

依赖关系

内部依赖

• langchain_core.runnables: Runnable 协议
• langchain_core.prompts: 提示模板
• langchain_core.memory: 记忆管理
• langchain_core.callbacks: 回调系统
• langchain.output_parsers: 输出解析器

外部依赖

• typing: 类型提示
• asyncio: 异步支持
• pydantic: 数据验证

数据流

LLMChain 流程

SequentialChain 流程

链类型详解

LLMChain

最基础的链，组合提示模板和 LLM：

from langchain.chains import LLMChain
from langchain_openai import OpenAI
from langchain_core.prompts import PromptTemplate

llm = OpenAI(temperature=0.9)
prompt = PromptTemplate(
    input_variables=["product"],
    template="What is a good name for a company that makes {product}?"
)
chain = LLMChain(llm=llm, prompt=prompt)

SimpleSequentialChain

简单顺序链，只有一个输入和输出：

from langchain.chains import LLMChain, SimpleSequentialChain

# 第一个链：生成公司名称
first_prompt = PromptTemplate.from_template(
    "What is a good name for a company that makes {product}?"
)
chain_one = LLMChain(llm=llm, prompt=first_prompt)

# 第二个链：生成公司描述
second_prompt = PromptTemplate.from_template(
    "Write a 20 words description for the following company:{company_name}"
)
chain_two = LLMChain(llm=llm, prompt=second_prompt)

# 组合成顺序链
overall_simple_chain = SimpleSequentialChain(
    chains=[chain_one, chain_two],
    verbose=True
)

SequentialChain

更复杂的顺序链，支持多个输入输出：

RouterChain

根据输入动态路由到不同子链：

from langchain.chains.router import MultiPromptChain
from langchain.prompts import PromptTemplate

# 定义多个提示模板
physics_template = """You are a very smart physics professor. \
You are great at answering questions about physics in a concise and easy to understand manner. \
When you don't know the answer to a question you admit that you don't know.

Here is a question:
{input}"""

math_template = """You are a very good mathematician. You are great at answering math questions. \
You are so good because you are able to break down hard problems into their component parts, \
answer the component parts, and then put them together to answer the broader question.

Here is a question:
{input}"""

# 创建路由链
chain = MultiPromptChain.from_prompts(
    llm=llm,
    prompts=[physics_template, math_template],
    names=["physics", "math"],
    description="Ask questions about physics or math",
)

RetrievalQA

检索增强生成（RAG）链：

链的组合方式

使用 LCEL 管道

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 现代推荐方式：LCEL
chain = (
    ChatPromptTemplate.from_template("Tell me a joke about {topic}")
    | ChatOpenAI(model="gpt-4")
    | StrOutputParser()
)

使用传统 Chain 类

from langchain.chains import LLMChain

# 传统方式（仍支持）
chain = LLMChain(
    llm=ChatOpenAI(model="gpt-4"),
    prompt=ChatPromptTemplate.from_template("Tell me a joke about {topic}")
)

记忆集成

添加记忆到链

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 带记忆的对话链
memory = ConversationBufferMemory()
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

记忆流程

输出解析

使用输出解析器

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

# 字符串输出
chain = prompt | llm | StrOutputParser()

# JSON 输出
json_parser = JsonOutputParser()
chain = prompt | llm | json_parser

自定义输出解析器

from langchain_core.output_parsers import BaseOutputParser

class CommaSeparatedListOutputParser(BaseOutputParser):
    def parse(self, text: str):
        return [item.strip() for item in text.split(",")]

链的执行模式

同步执行

result = chain.invoke({"input": "Hello"})

异步执行

result = await chain.ainvoke({"input": "Hello"})

流式执行

for chunk in chain.stream({"input": "Hello"}):
    print(chunk, end="")

批量执行

results = chain.batch([
    {"input": "Hello"},
    {"input": "World"}
])

链的调试

使用 Verbose 模式

chain = LLMChain(llm=llm, prompt=prompt, verbose=True)

使用 LangSmith 追踪

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"

# 所有执行会自动追踪
result = chain.invoke({"input": "Hello"})

使用回调

from langchain_core.callbacks import StdOutCallbackHandler

result = chain.invoke(
    {"input": "Hello"},
    config={"callbacks": [StdOutCallbackHandler()]}
)

链的最佳实践

1. 使用 LCEL 而非传统链

# 推荐：LCEL
chain = prompt | llm | parser

# 不推荐：传统方式
chain = LLMChain(llm=llm, prompt=prompt)

2. 明确输入输出模式

from langchain_core.pydantic_v1 import BaseModel, Field

class ChainInput(BaseModel):
    topic: str = Field(..., description="The topic to write about")

class ChainOutput(BaseModel):
    story: str = Field(..., description="The generated story")

chain = prompt | llm | parser

3. 使用类型提示

from typing import Dict, List

def process(inputs: Dict[str, str]) -> List[str]:
    # 类型提示帮助 IDE 和文档
    return chain.invoke(inputs)

迁移到 LangGraph

LangGraph 提供比传统链更强大的功能：

• 状态持久化
• 复杂控制流（循环、分支）
• 人机协作
• 时间旅行调试

---