LangChain Chains 链

Chains 是 LangChain 的核心概念，让你可以将多个组件串联起来，构建复杂的 LLM 工作流。

#什么是 Chain？

Chain 是将多个组件（LLM、Prompt、Parser 等）组合在一起的方式：

输入 → Prompt → LLM → Parser → 输出

#LCEL 基础

#管道操作符 |

python复制
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 组件
prompt = ChatPromptTemplate.from_template("将 {text} 翻译成 {language}")
llm = ChatOpenAI()
parser = StrOutputParser()

# 使用 | 组合
chain = prompt | llm | parser

# 调用
result = chain.invoke({
    "text": "Hello, world!",
    "language": "中文"
})
print(result)  # 你好，世界！

#链的方法

python复制
# 单次调用
result = chain.invoke({"text": "Hello"})

# 流式输出
for chunk in chain.stream({"text": "Hello"}):
    print(chunk, end="")

# 批量处理
results = chain.batch([
    {"text": "Hello"},
    {"text": "Goodbye"}
])

# 异步调用
result = await chain.ainvoke({"text": "Hello"})

#常用 Chain 模式

#顺序链

python复制
# 翻译 → 总结
translate_prompt = ChatPromptTemplate.from_template(
    "将以下文本翻译成英文：\n{text}"
)

summarize_prompt = ChatPromptTemplate.from_template(
    "用一句话总结：\n{translated}"
)

# 组合
translate_chain = translate_prompt | llm | StrOutputParser()
summarize_chain = summarize_prompt | llm | StrOutputParser()

# 顺序执行
full_chain = (
    {"translated": translate_chain}
    | summarize_chain
)

result = full_chain.invoke({"text": "这是一段很长的中文文本..."})

#并行链

python复制
from langchain_core.runnables import RunnableParallel

# 同时执行多个任务
parallel_chain = RunnableParallel(
    summary=ChatPromptTemplate.from_template("总结：{text}") | llm | StrOutputParser(),
    keywords=ChatPromptTemplate.from_template("提取关键词：{text}") | llm | StrOutputParser(),
    sentiment=ChatPromptTemplate.from_template("分析情感：{text}") | llm | StrOutputParser()
)

result = parallel_chain.invoke({"text": "这是一篇关于人工智能的文章..."})
print(result["summary"])
print(result["keywords"])
print(result["sentiment"])

#条件链

python复制
from langchain_core.runnables import RunnableBranch

# 根据条件选择不同的处理路径
branch_chain = RunnableBranch(
    # (条件, 执行的链)
    (lambda x: x["type"] == "code", code_review_chain),
    (lambda x: x["type"] == "text", text_analysis_chain),
    # 默认分支
    default_chain
)

result = branch_chain.invoke({"type": "code", "content": "def hello(): pass"})

#传递数据

python复制
from langchain_core.runnables import RunnablePassthrough

# RunnablePassthrough 传递原始输入
chain = (
    {
        "original": RunnablePassthrough(),  # 保留原始输入
        "processed": some_processing_chain   # 处理后的结果
    }
    | final_chain
)

#实用 Chain 示例

#问答 Chain

python复制
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。基于以下上下文回答问题。\n\n上下文：{context}"),
    ("human", "{question}")
])

qa_chain = prompt | ChatOpenAI() | StrOutputParser()

result = qa_chain.invoke({
    "context": "LangChain 是一个用于构建 LLM 应用的框架...",
    "question": "LangChain 是什么？"
})

#多步推理 Chain

python复制
# 步骤 1: 理解问题
understand_prompt = ChatPromptTemplate.from_template(
    "分析这个问题的关键点：{question}"
)

# 步骤 2: 查找相关信息
search_prompt = ChatPromptTemplate.from_template(
    "基于分析 '{analysis}'，我需要查找什么信息？"
)

# 步骤 3: 生成答案
answer_prompt = ChatPromptTemplate.from_template(
    "问题：{question}\n分析：{analysis}\n信息：{info}\n\n请回答问题。"
)

# 组合
reasoning_chain = (
    {"analysis": understand_prompt | llm | StrOutputParser(), "question": RunnablePassthrough()}
    | {"info": search_prompt | llm | StrOutputParser(), "analysis": lambda x: x["analysis"], "question": lambda x: x["question"]}
    | answer_prompt | llm | StrOutputParser()
)

#带验证的 Chain

python复制
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, validator

class Answer(BaseModel):
    answer: str
    confidence: float

    @validator('confidence')
    def check_confidence(cls, v):
        if not 0 <= v <= 1:
            raise ValueError('Confidence must be between 0 and 1')
        return v

parser = JsonOutputParser(pydantic_object=Answer)

chain = (
    prompt
    | llm
    | parser  # 自动验证输出格式
)

#带重试的 Chain

python复制
from langchain_core.runnables import RunnableWithFallbacks

# 主 Chain
main_chain = prompt | ChatOpenAI(model="gpt-4o") | StrOutputParser()

# 备用 Chain
fallback_chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

# 带回退的 Chain
robust_chain = main_chain.with_fallbacks([fallback_chain])

#调试和监控

#添加日志

python复制
from langchain_core.runnables import RunnableLambda

def log_input(x):
    print(f"输入: {x}")
    return x

def log_output(x):
    print(f"输出: {x}")
    return x

chain = (
    RunnableLambda(log_input)
    | prompt
    | llm
    | parser
    | RunnableLambda(log_output)
)

#使用 LangSmith

python复制
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

# 所有 Chain 调用都会被追踪
result = chain.invoke({"text": "Hello"})

#自定义组件

#自定义 Runnable

python复制
from langchain_core.runnables import RunnableLambda

def custom_transform(text: str) -> str:
    """自定义转换函数"""
    return text.upper()

# 转换为 Runnable
custom_runnable = RunnableLambda(custom_transform)

# 在 Chain 中使用
chain = prompt | llm | StrOutputParser() | custom_runnable

#异步自定义组件

python复制
async def async_transform(text: str) -> str:
    # 异步操作
    await some_async_operation()
    return text

custom_async = RunnableLambda(func=sync_transform, afunc=async_transform)

#最佳实践

#1. 保持 Chain 简单

python复制
# ❌ 过于复杂的单个 Chain
mega_chain = step1 | step2 | step3 | step4 | step5 | step6

# ✅ 分解为多个可复用的 Chain
preprocess_chain = step1 | step2
process_chain = step3 | step4
postprocess_chain = step5 | step6

full_chain = preprocess_chain | process_chain | postprocess_chain

#2. 错误处理

python复制
from langchain_core.runnables import RunnableWithFallbacks

chain = main_chain.with_fallbacks(
    [fallback_chain],
    exceptions_to_handle=(Exception,)
)

#3. 类型安全

python复制
from langchain_core.runnables import RunnableSerializable

class MyChain(RunnableSerializable[dict, str]):
    """类型化的自定义 Chain"""

    def invoke(self, input: dict) -> str:
        # 实现逻辑
        return "result"

#下一步

• Agents 代理 - 构建智能代理
• RAG 检索增强 - 知识库问答
• LangGraph - 复杂工作流

---

提示：LCEL 是 LangChain 的推荐方式，比旧版 Chain 类更灵活、更高效。