在大语言模型应用开发中，LangChain作为最流行的框架之一，其核心概念Chain和Agent的理解至关重要。本文将深入解析这两个概念，通过丰富的代码示例和实际应用场景，帮助你掌握LangChain的核心机制。

1. 引言：为什么需要LangChain？

随着大语言模型的快速发展，我们面临着新的挑战：如何将LLM与外部工具、数据源和计算资源有效结合？LangChain应运而生，它提供了一个统一的框架来构建基于LLM的应用程序。

1.1 LangChain的核心价值

• 模块化设计：将复杂任务分解为可重用的组件
• 链式调用：将多个组件串联成完整的工作流
• 工具集成：让LLM能够使用外部工具和API
• 状态管理：维护对话历史和上下文信息

2. Chain（链）概念深度解析

2.1 什么是Chain？

Chain是LangChain中最基本的概念，它代表一系列组件的顺序执行。可以将Chain理解为LLM应用的"工作流"或"管道"。

Chain的核心特征：

• 顺序执行：组件按预定顺序执行
• 数据传递：前一个组件的输出作为后一个组件的输入
• 确定性：执行路径在运行时是确定的
• 可组合性：Chain本身可以作为其他Chain的组件

2.2 Chain的工作原理

让我们通过一个流程图来理解Chain的基本工作原理：

2.3 基础Chain类型详解

2.3.1 LLMChain - 最基础的Chain

LLMChain结合了LLM、提示模板和输出解析器，是最常用的Chain类型。

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.schema import BaseOutputParser
import os

# 设置OpenAI API密钥
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# 自定义输出解析器
class CommaSeparatedListOutputParser(BaseOutputParser):
    """将输出解析为逗号分隔的列表"""
    
    def parse(self, text: str):
        """解析LLM的输出"""
        return [item.strip() for item in text.split(",") if item.strip()]

# 创建LLM实例
llm = OpenAI(temperature=0.7)

# 创建提示模板
prompt_template = PromptTemplate(
    input_variables=["topic"],
    template="请生成关于{topic}的5个相关关键词，用逗号分隔："
)

# 创建LLMChain
llm_chain = LLMChain(
    llm=llm,
    prompt=prompt_template,
    output_parser=CommaSeparatedListOutputParser()
)

# 使用Chain
try:
    result = llm_chain.run("人工智能")
    print("生成的关键词:", result)
except Exception as e:
    print(f"执行出错: {e}")

2.3.2 SequentialChain - 顺序执行链

SequentialChain允许我们将多个Chain连接起来，形成复杂的工作流。

from langchain.chains import SimpleSequentialChain, TransformChain
import re

# 第一个Chain：生成文章大纲
outline_prompt = PromptTemplate(
    input_variables=["topic"],
    template="请为关于{topic}的文章生成一个详细大纲，包含主要章节和子主题："
)
outline_chain = LLMChain(llm=llm, prompt=outline_prompt)

# 转换Chain：清理和格式化文本
def transform_func(inputs):
    text = inputs["text"]
    # 移除多余的空行和空格
    cleaned_text = re.sub(r'\n\s*\n', '\n\n', text)
    return {"cleaned_text": cleaned_text}

transform_chain = TransformChain(
    input_variables=["text"],
    output_variables=["cleaned_text"],
    transform=transform_func
)

# 第二个Chain：基于大纲写文章
article_prompt = PromptTemplate(
    input_variables=["cleaned_text"],
    template="根据以下大纲撰写一篇完整的文章：\n{cleaned_text}\n\n文章内容："
)
article_chain = LLMChain(llm=llm, prompt=article_prompt)

# 创建顺序链
full_chain = SimpleSequentialChain(
    chains=[outline_chain, transform_chain, article_chain],
    verbose=True  # 显示详细执行过程
)

# 执行完整的工作流
try:
    final_result = full_chain.run("机器学习在医疗诊断中的应用")
    print("最终生成的文章:")
    print(final_result)
except Exception as e:
    print(f"链执行出错: {e}")

2.3.3 RouterChain - 路由链

RouterChain根据输入内容决定执行哪个子Chain，实现条件逻辑。

from langchain.chains.router import MultiRouteChain
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE

# 定义不同领域的提示模板
prompt_infos = [
    {
        "name": "technology",
        "description": "适合技术相关的问题",
        "prompt_template": "你是一个技术专家。请用专业的技术语言回答：{input}"
    },
    {
        "name": "business", 
        "description": "适合商业和市场相关的问题",
        "prompt_template": "你是一个商业顾问。请从商业角度分析：{input}"
    },
    {
        "name": "general",
        "description": "适合一般性问题",
        "prompt_template": "请以友好的方式回答这个问题：{input}"
    }
]

# 创建各个领域的Chain
destination_chains = {}
for p_info in prompt_infos:
    prompt = PromptTemplate(
        template=p_info["prompt_template"],
        input_variables=["input"]
    )
    destination_chains[p_info["name"]] = LLMChain(llm=llm, prompt=prompt)

# 默认Chain
default_chain = LLMChain(llm=llm, prompt=PromptTemplate(
    template="请回答这个问题：{input}",
    input_variables=["input"]
))

# 创建路由Chain
router_chain = LLMRouterChain.from_llm(
    llm, 
    MULTI_PROMPT_ROUTER_TEMPLATE.format(
        destinations="\n".join([f"{p['name']}: {p['description']}" for p in prompt_infos])
    )
)

# 创建多路由Chain
multi_chain = MultiRouteChain(
    router_chain=router_chain,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True
)

# 测试路由功能
test_questions = [
    "解释Transformer架构的工作原理",
    "如何制定数字营销策略",
    "今天天气怎么样"
]

for question in test_questions:
    print(f"\n问题: {question}")
    print("回答:", multi_chain.run(question))

3. Agent（代理）概念深度解析

3.1 什么是Agent？

Agent是更高级的LangChain组件，它让LLM能够自主决定使用哪些工具来完成任务。与Chain的确定性不同，Agent具有决策能力。

Agent的核心特征：

• 自主决策：根据当前状态决定下一步行动
• 工具使用：能够调用外部工具和API
• 迭代执行：通过思考-行动-观察的循环逐步推进
• 不确定性：执行路径在运行时动态决定

3.2 Agent的工作原理

Agent的工作流程可以通过以下状态图表示：

3.3 Agent的核心组件

3.3.1 工具（Tools）

工具是Agent可以调用的外部功能，可以是API、函数、或其他服务。

from langchain.agents import Tool, AgentExecutor, LLMSingleActionAgent
from langchain import SerpAPIWrapper
from datetime import datetime
import math
import requests

# 自定义工具：计算器
def calculator(query):
    """执行数学计算"""
    try:
        # 移除中文描述，只保留数学表达式
        expression = query.replace("计算", "").replace("算一下", "").strip()
        # 安全评估数学表达式
        allowed_chars = set('0123456789+-*/(). ')
        if all(c in allowed_chars for c in expression):
            result = eval(expression)
            return f"计算结果: {expression} = {result}"
        else:
            return "错误: 包含不安全的字符"
    except Exception as e:
        return f"计算错误: {e}"

# 自定义工具：时间查询
def get_current_time(query):
    """获取当前时间"""
    now = datetime.now()
    return f"当前时间: {now.strftime('%Y-%m-%d %H:%M:%S')}"

# 自定义工具：天气查询（模拟）
def get_weather(city):
    """获取城市天气信息"""
    # 这里使用模拟数据，实际应用中可接入天气API
    weather_data = {
        "北京": "晴，15°C",
        "上海": "多云，18°C", 
        "深圳": "雨，22°C",
        "成都": "阴，16°C"
    }
    return weather_data.get(city, f"未找到{city}的天气信息")

# 创建工具列表
tools = [
    Tool(
        name="Calculator",
        func=calculator,
        description="用于数学计算。输入数学表达式如'2+2'或'3*5'"
    ),
    Tool(
        name="Time",
        func=get_current_time,
        description="获取当前日期和时间"
    ),
    Tool(
        name="Weather",
        func=get_weather,
        description="查询城市天气。输入城市名称如'北京'或'上海'"
    )
]

3.3.2 代理类型详解

ReAct代理 - 最常用的代理类型

ReAct代理通过Reasoning和Acting的循环来解决问题。

from langchain.agents import initialize_agent
from langchain.agents import AgentType

# 初始化ReAct代理
react_agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True
)

# 测试复杂任务
complex_queries = [
    "先计算一下(15 * 4) + (20 / 2)等于多少，然后告诉我现在的时间",
    "查询北京的天气，如果是晴天就计算25的平方根",
    "现在的时间是什么？如果时间是下午，就计算从2020到2024年有多少年"
]

for i, query in enumerate(complex_queries, 1):
    print(f"\n{'='*50}")
    print(f"任务 {i}: {query}")
    print(f"{'='*50}")
    try:
        result = react_agent.run(query)
        print(f"结果: {result}")
    except Exception as e:
        print(f"执行失败: {e}")

自定义代理

我们可以创建更专业的自定义代理来处理特定领域的任务。

from langchain.agents import Tool, AgentExecutor
from langchain.schema import SystemMessage
from langchain.prompts import BaseChatPromptTemplate
from typing import List, Union
from langchain.schema import HumanMessage, AIMessage

# 自定义提示模板
class CustomPromptTemplate(BaseChatPromptTemplate):
    template: str
    
    def format_messages(self, **kwargs) -> str:
        # 格式化提示信息
        intermediate_steps = kwargs.pop("intermediate_steps")
        thoughts = ""
        
        for action, observation in intermediate_steps:
            thoughts += f"行动: {action.log}\n"
            thoughts += f"观察: {observation}\n"
            
        kwargs["agent_scratchpad"] = thoughts
        formatted_template = self.template.format(**kwargs)
        
        return [HumanMessage(content=formatted_template)]

# 创建自定义代理
custom_prompt = CustomPromptTemplate(
    template="""你是一个智能助手，可以访问各种工具来帮助用户。

你可以使用的工具:
{tools}

使用以下格式:
问题: 用户输入的问题
思考: 你需要思考如何逐步解决问题
行动: 要使用的工具名称
行动输入: 工具的输入参数
观察: 工具返回的结果
... (这个思考/行动/观察的循环可以重复多次)
思考: 我现在知道最终答案了
最终答案: 对原始问题的最终回答

开始!

问题: {input}
{agent_scratchpad}"""
)

# 测试自定义代理
from langchain.agents import LLMSingleActionAgent

custom_agent = LLMSingleActionAgent(
    llm_chain=LLMChain(llm=llm, prompt=custom_prompt),
    output_parser=ReactSingleInputOutputParser(),
    stop=["\n观察:"],
    allowed_tools=[tool.name for tool in tools]
)

agent_executor = AgentExecutor.from_agent_and_tools(
    agent=custom_agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True
)

# 测试自定义代理
test_query = "先告诉我现在的时间，然后计算当前时间的小时数乘以60是多少"
result = agent_executor.run(test_query)
print(f"自定义代理结果: {result}")

4. Chain vs Agent：应用场景对比

4.1 何时使用Chain？

Chain适合确定性、可预测的工作流。

Chain的典型应用场景：

# 场景1：文档处理流水线
from langchain.chains import RetrievalQA
from langchain.document_loaders import TextLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

def create_document_qa_chain(file_path):
    """创建文档问答Chain"""
    # 加载文档
    loader = TextLoader(file_path, encoding='utf-8')
    documents = loader.load()
    
    # 创建向量数据库
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma.from_documents(documents, embeddings)
    
    # 创建检索Chain
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vectorstore.as_retriever(),
        return_source_documents=True
    )
    
    return qa_chain

# 场景2：内容生成流水线
def create_content_generation_chain():
    """创建内容生成Chain"""
    # 主题生成
    topic_prompt = PromptTemplate(
        input_variables=["industry"],
        template="为{industry}行业生成5个内容营销主题"
    )
    topic_chain = LLMChain(llm=llm, prompt=topic_prompt)
    
    # 大纲生成
    outline_prompt = PromptTemplate(
        input_variables=["topic"],
        template="为'{topic}'创建详细的内容大纲"
    )
    outline_chain = LLMChain(llm=llm, prompt=outline_prompt)
    
    # 内容生成
    content_prompt = PromptTemplate(
        input_variables=["outline"],
        template="根据以下大纲撰写完整内容：\n{outline}"
    )
    content_chain = LLMChain(llm=llm, prompt=content_prompt)
    
    # 组合成顺序链
    from langchain.chains import SequentialChain
    
    full_chain = SequentialChain(
        chains=[topic_chain, outline_chain, content_chain],
        input_variables=["industry"],
        output_variables=["text"],
        verbose=True
    )
    
    return full_chain

4.2 何时使用Agent？

Agent适合需要动态决策、工具选择的复杂任务。

Agent的典型应用场景：

# 场景1：研究助手Agent
def create_research_agent():
    """创建研究助手Agent"""
    
    # 研究工具
    def search_research_papers(query):
        """搜索研究论文（模拟）"""
        return f"找到关于'{query}'的3篇相关论文: 1. 论文A, 2. 论文B, 3. 论文C"
    
    def summarize_content(content):
        """总结内容"""
        summary_prompt = PromptTemplate(
            input_variables=["content"],
            template="请总结以下内容：\n{content}"
        )
        summary_chain = LLMChain(llm=llm, prompt=summary_prompt)
        return summary_chain.run(content=content)
    
    def compare_concepts(concepts):
        """比较概念"""
        compare_prompt = PromptTemplate(
            input_variables=["concepts"],
            template="请比较以下概念的异同：{concepts}"
        )
        compare_chain = LLMChain(llm=llm, prompt=compare_prompt)
        return compare_chain.run(concepts=concepts)
    
    research_tools = [
        Tool(name="SearchPapers", func=search_research_papers, 
             description="搜索学术论文"),
        Tool(name="Summarize", func=summarize_content,
             description="总结文本内容"),
        Tool(name="Compare", func=compare_concepts,
             description="比较不同概念")
    ]
    
    research_agent = initialize_agent(
        tools=research_tools,
        llm=llm,
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True
    )
    
    return research_agent

# 测试研究助手
research_agent = create_research_agent()
research_query = """
研究机器学习中的Transformer架构：
1. 先搜索相关的论文
2. 总结找到的主要内容
3. 比较Transformer与RNN的异同
"""
result = research_agent.run(research_query)
print(result)

5. 高级应用：Chain与Agent的结合

在实际应用中，我们经常需要将Chain和Agent结合起来使用。

5.1 混合架构示例

from langchain.agents import AgentType, Tool, initialize_agent
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

def create_hybrid_system():
    """创建Chain和Agent结合的混合系统"""
    
    # 1. 创建对话Chain（有记忆功能）
    memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
    conversation_chain = ConversationChain(llm=llm, memory=memory, verbose=True)
    
    # 2. 创建工具
    def handle_conversation(query):
        """处理对话的工具"""
        return conversation_chain.run(input=query)
    
    # 3. 将Chain作为工具提供给Agent
    hybrid_tools = [
        Tool(name="Chat", func=handle_conversation,
             description="用于日常对话和上下文交流"),
        Tool(name="Calculator", func=calculator,
             description="用于数学计算"),
        Tool(name="Time", func=get_current_time,
             description="获取当前时间")
    ]
    
    # 4. 创建智能Agent
    hybrid_agent = initialize_agent(
        tools=hybrid_tools,
        llm=llm,
        agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
        verbose=True,
        memory=memory
    )
    
    return hybrid_agent

# 测试混合系统
hybrid_system = create_hybrid_system()

# 模拟对话
conversation = [
    "你好，我是小明",
    "请计算一下25的平方根",
    "还记得我的名字吗？",
    "现在几点了？"
]

for message in conversation:
    print(f"\n用户: {message}")
    response = hybrid_system.run(input=message)
    print(f"助手: {response}")

6. 性能优化和最佳实践

6.1 Chain优化技巧

# 1. 缓存优化
from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# 启用缓存
set_llm_cache(InMemoryCache())

# 2. 批量处理
def batch_process_chain(chain, inputs):
    """批量处理Chain任务"""
    results = []
    for input_text in inputs:
        result = chain.run(input_text)
        results.append(result)
    return results

# 3. 错误处理和重试
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def robust_chain_execution(chain, input_data):
    """带重试机制的Chain执行"""
    try:
        return chain.run(input_data)
    except Exception as e:
        print(f"Chain执行失败: {e}")
        raise

6.2 Agent优化策略

# 1. 工具优化
def create_optimized_agent():
    """创建优化后的Agent"""
    
    # 工具描述优化
    optimized_tools = [
        Tool(
            name="PreciseCalculator",
            func=calculator,
            description="执行数学计算。输入应该是纯数学表达式，如：(15 + 3) * 2"
        ),
        Tool(
            name="AccurateTime",
            func=get_current_time, 
            description="返回当前的精确日期和时间，格式为YYYY-MM-DD HH:MM:SS"
        )
    ]
    
    # 使用更高效的代理类型
    optimized_agent = initialize_agent(
        tools=optimized_tools,
        llm=llm,
        agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True,
        max_iterations=5,  # 限制最大迭代次数
        early_stopping_method="generate"  # 提前停止策略
    )
    
    return optimized_agent

7. 总结

7.1 核心要点回顾

Chain的特点：

• 确定性执行流程
• 适合结构化任务
• 性能可预测
• 易于调试和测试

Agent的特点：

• 动态决策能力
• 适合复杂、开放性问题
• 能够使用外部工具
• 更具灵活性

7.2 选择指南

场景特征	推荐选择	理由
固定工作流	Chain	执行路径确定，性能稳定
需要工具选择	Agent	动态决策使用哪个工具
简单数据处理	Chain	结构清晰，易于维护
复杂问题解决	Agent	需要多步推理和工具使用
实时性要求高	Chain	执行时间可预测
探索性任务	Agent	路径不确定，需要灵活性

7.3 实践建议

1. 从Chain开始：对于大多数业务场景，先尝试用Chain解决
2. 渐进式复杂化：当Chain无法满足时再引入Agent
3. 混合使用：在复杂系统中结合Chain和Agent的优势
4. 监控和评估：建立完善的监控体系来评估系统性能

通过本文的详细讲解和丰富示例，相信你已经对LangChain中的Chain和Agent有了深入的理解。在实际项目中，根据具体需求灵活选择和组合这些组件，将帮助你构建出更加强大和智能的LLM应用。

---

欢迎在评论区分享你的LangChain使用经验和遇到的问题！ 🚀