跳至主要內容

LangChain 深度实战——Chain、Agent、Memory 核心组件完整使用指南

老张2026/4/30大约 11 分钟

LangChain 深度实战——Chain、Agent、Memory 核心组件完整使用指南

适读人群：有 Python 基础的 Java 工程师、AI 应用开发者 | 阅读时长：约18分钟 | 核心价值：彻底搞懂 LangChain 三大核心组件，能独立搭建生产级 AI 应用

老王是个工作了八年的 Java 架构师，去年公司战略转型，要做一个内部知识库问答系统。他信心满满地接下来，觉得自己搞过 Spring 全家桶，什么框架没见过，LangChain 能有多难？

然后他就掉进去了。

第一周，他把 LangChain 的官方文档从头到尾啃了一遍，发现概念一堆：Chain、Agent、Memory、Tool、Retriever……每个词单独看都能理解，但组合起来就搞不清楚谁调用谁。他写了个最简单的问答链，跑通了，但一加上记忆功能，上下文就乱掉，前三轮的对话内容直接被截断，用户体验极差。

第二周，他开始研究 Agent，想让 AI 自动决定要不要去数据库查数据。结果 Agent 每次运行要花十几秒，偶尔还死循环，日志里出现了 "I need to use the tool again..." 的无限递归。他不知道是模型的问题还是代码的问题，一度怀疑人生。

他找到我的时候，语气有点颓："老张，我感觉 LangChain 文档写得太碎了，东西很多，但就是不知道在实际项目里怎么用。"

我说：你的问题不是不会用，是没有建立起这三个组件的心智模型。Chain 是流水线，Memory 是状态机，Agent 是决策树。搞清楚这三个关系，LangChain 就通了。

今天我们就系统地把这三块讲透，给你一套能直接用的工程实践方案。

一、LangChain 的核心设计哲学

在学具体 API 之前，有必要理解 LangChain 为什么这么设计。

LangChain 的本质是把 LLM 调用流程模块化、可组合化。传统 API 调用是一次性的：输入 → 模型 → 输出。但现实中的 AI 应用远比这复杂：你可能需要先检索文档，再传给模型；需要多轮对话保持上下文；需要让模型自主决定要不要调用外部工具。

LangChain 用三个核心抽象来解决这些问题：

  • Chain（链）：把多个步骤串起来，每步的输出是下步的输入
  • Memory（记忆）：让对话有状态，跨轮次保持上下文
  • Agent（智能体）：让模型自主决策，动态选择下一步动作

二、Chain 深度实战

2.1 从最简单的 LLMChain 开始

import os
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain

# 初始化模型，这里用 gpt-3.5-turbo
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,
    openai_api_key=os.environ.get("OPENAI_API_KEY")
)

# 定义提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的代码审查专家，擅长发现潜在的 Bug 和性能问题。"),
    ("human", "请审查以下 {language} 代码，指出问题并给出改进建议：\n\n{code}")
])

# 创建 Chain
code_review_chain = LLMChain(
    llm=llm,
    prompt=prompt,
    output_key="review_result"
)

# 运行
result = code_review_chain.run({
    "language": "Python",
    "code": """
def get_user_data(user_ids):
    results = []
    for uid in user_ids:
        data = db.query(f"SELECT * FROM users WHERE id = {uid}")
        results.append(data)
    return results
"""
})

print(result)

这段代码有一个经典的 SQL 注入漏洞，模型会直接指出来。LLMChain 是最基础的 Chain，一个输入模板加一个模型调用。

2.2 SequentialChain——多步串联

真实业务场景里，你往往需要多步处理。比如：先提取文档关键信息，再基于关键信息生成摘要，再基于摘要生成社交媒体帖子。

from langchain.chains import SequentialChain

# 第一步：提取关键信息
extract_prompt = ChatPromptTemplate.from_template(
    "从以下技术文档中提取3-5个核心技术要点，每点一行：\n\n{document}"
)
extract_chain = LLMChain(
    llm=llm,
    prompt=extract_prompt,
    output_key="key_points"
)

# 第二步：生成摘要
summary_prompt = ChatPromptTemplate.from_template(
    "基于以下技术要点，写一段200字以内的技术摘要：\n\n{key_points}"
)
summary_chain = LLMChain(
    llm=llm,
    prompt=summary_prompt,
    output_key="summary"
)

# 第三步：生成朋友圈文案
social_prompt = ChatPromptTemplate.from_template(
    "把以下技术摘要改写成适合技术人员朋友圈的文案，语气轻松专业，加上2-3个相关emoji：\n\n{summary}"
)
social_chain = LLMChain(
    llm=llm,
    prompt=social_prompt,
    output_key="social_post"
)

# 串联三个 Chain
overall_chain = SequentialChain(
    chains=[extract_chain, summary_chain, social_chain],
    input_variables=["document"],
    output_variables=["key_points", "summary", "social_post"],
    verbose=True  # 开启详细日志，方便调试
)

# 实际调用
document = """
LangChain v0.2 对 LCEL（LangChain Expression Language）做了重大升级，
引入了更灵活的 Runnable 接口，支持并行执行多个 Chain，
内存管理也优化为基于消息列表的方式，与 OpenAI 的消息格式完全对齐……
"""

result = overall_chain({"document": document})
print("关键要点：", result["key_points"])
print("摘要：", result["summary"])
print("朋友圈文案：", result["social_post"])

2.3 踩坑实录一：SequentialChain 变量名冲突

现象：运行 SequentialChain 时报错 Missing required input variables，或者中间某步的输出没有传递到下一步。

原因：每个 Chain 的 output_key 必须唯一，且下一个 Chain 的 prompt 模板中的变量名必须与上一个 Chain 的 output_key 完全匹配，大小写敏感。

解法：

# 错误示例：output_key 和 prompt 变量名不一致
extract_chain = LLMChain(output_key="keyPoints")  # 驼峰
summary_prompt = ChatPromptTemplate.from_template("{key_points}...")  # 下划线

# 正确做法：保持一致，推荐全用下划线
extract_chain = LLMChain(output_key="key_points")
summary_prompt = ChatPromptTemplate.from_template("{key_points}...")

三、Memory 深度实战

3.1 Memory 的类型选择

LangChain 提供了多种 Memory 类型，每种适合不同场景：

Memory 类型特点适用场景
ConversationBufferMemory保存完整对话历史短对话，对话轮次少
ConversationBufferWindowMemory只保留最近 K 轮中等长度对话
ConversationSummaryMemory用 LLM 压缩历史长对话，节省 Token
ConversationTokenBufferMemory按 Token 数截断精确控制成本

3.2 完整的多轮对话实现

from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 只保留最近5轮对话，避免 context 过长
memory = ConversationBufferWindowMemory(
    k=5,
    return_messages=True,  # 返回消息列表格式，而不是字符串
    memory_key="chat_history",
    human_prefix="用户",
    ai_prefix="AI助手"
)

# 自定义 Prompt，加入系统人设
prompt = PromptTemplate(
    input_variables=["chat_history", "input"],
    template="""你是一个专业的 Java 转 Python 学习助手，帮助 Java 工程师快速上手 Python。
你的回答风格：简洁直接，多给代码示例，遇到 Java 和 Python 的差异要主动对比说明。

历史对话：
{chat_history}

用户：{input}
AI助手："""
)

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    prompt=prompt,
    verbose=False
)

# 模拟多轮对话
turns = [
    "Python 的列表推导式怎么用？",
    "那和 Java 的 Stream API 有什么区别？",
    "能给我一个实际的例子吗，比如过滤大于100的数字？",
]

for user_input in turns:
    print(f"\n用户：{user_input}")
    response = conversation.predict(input=user_input)
    print(f"AI助手：{response}")

# 查看当前内存状态
print("\n当前记忆内容：")
print(memory.load_memory_variables({}))

3.3 踩坑实录二：Memory 在多用户场景下共享状态

现象：你的聊天机器人服务上线后，用户 A 的对话历史混入了用户 B 的上下文，出现了"我没说过这个"的情况。

原因：Memory 对象是有状态的，如果你在 Flask/FastAPI 里把它定义为全局变量或单例，所有请求会共享同一个 Memory 实例。

解法：每个会话（session）创建独立的 Memory 实例，用 session_id 管理：

from langchain.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories import ChatMessageHistory
from typing import Dict

# 存储每个 session 的对话历史
session_store: Dict[str, ChatMessageHistory] = {}

def get_memory_for_session(session_id: str) -> ConversationBufferWindowMemory:
    """为每个 session 返回独立的 Memory 实例"""
    if session_id not in session_store:
        session_store[session_id] = ChatMessageHistory()
    
    return ConversationBufferWindowMemory(
        k=10,
        chat_memory=session_store[session_id],
        return_messages=True,
        memory_key="chat_history"
    )

# FastAPI 路由示例
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    session_id: str
    message: str

@app.post("/chat")
async def chat(request: ChatRequest):
    memory = get_memory_for_session(request.session_id)
    conversation = ConversationChain(llm=llm, memory=memory)
    response = conversation.predict(input=request.message)
    return {"response": response}

四、Agent 深度实战

4.1 Agent 的核心机制

Agent 的工作原理是 ReAct（Reasoning + Acting）：模型先思考（Thought），然后决定执行什么动作（Action），看到动作结果（Observation），再继续思考，直到得出最终答案（Final Answer）。

理解这个循环非常重要，很多 Agent 调试问题都出在这里。

4.2 自定义 Tool + Agent 完整实现

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
import json
import requests
from datetime import datetime

# 定义工具集
@tool
def get_stock_price(ticker: str) -> str:
    """
    获取股票的当前价格信息。
    Args:
        ticker: 股票代码，如 AAPL, GOOGL
    Returns:
        包含股票价格信息的字符串
    """
    # 实际项目中调用真实 API，这里用模拟数据
    mock_prices = {
        "AAPL": {"price": 189.50, "change": "+1.2%", "volume": "52M"},
        "GOOGL": {"price": 142.30, "change": "-0.5%", "volume": "18M"},
        "MSFT": {"price": 415.20, "change": "+0.8%", "volume": "22M"},
    }
    
    ticker = ticker.upper()
    if ticker in mock_prices:
        data = mock_prices[ticker]
        return f"{ticker}: 价格 ${data['price']}, 涨跌 {data['change']}, 成交量 {data['volume']}"
    return f"未找到 {ticker} 的数据"


@tool
def calculate_portfolio_value(holdings: str) -> str:
    """
    计算投资组合总价值。
    Args:
        holdings: JSON 格式的持仓，如 '{"AAPL": 10, "GOOGL": 5}'
    Returns:
        投资组合总价值
    """
    mock_prices = {"AAPL": 189.50, "GOOGL": 142.30, "MSFT": 415.20}
    
    try:
        portfolio = json.loads(holdings)
        total = 0
        details = []
        for ticker, shares in portfolio.items():
            ticker = ticker.upper()
            if ticker in mock_prices:
                value = mock_prices[ticker] * shares
                total += value
                details.append(f"{ticker} x{shares} = ${value:.2f}")
        
        return f"持仓明细：{', '.join(details)}\n总价值：${total:.2f}"
    except json.JSONDecodeError:
        return "持仓格式错误，请使用 JSON 格式"


@tool
def get_market_news(topic: str) -> str:
    """
    获取最新市场新闻。
    Args:
        topic: 查询主题，如 AI、芯片、新能源
    Returns:
        相关新闻摘要
    """
    # 实际项目接入新闻 API
    news_db = {
        "AI": "OpenAI 发布 GPT-5，性能全面超越 GPT-4；英伟达 AI 芯片需求持续旺盛",
        "芯片": "台积电 2nm 工艺量产在即；英特尔宣布重组计划",
        "新能源": "特斯拉 Q3 交付量超预期；比亚迪海外扩张加速",
    }
    return news_db.get(topic, f"暂无关于 {topic} 的最新新闻")


# 创建 Agent
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

tools = [get_stock_price, calculate_portfolio_value, get_market_news]

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的投资助理，帮助用户分析投资组合。
你可以：
1. 查询个股价格
2. 计算投资组合价值
3. 获取市场新闻

回答时要简洁，数据要准确，必要时给出简短分析。"""),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,  # 重要：防止死循环
    handle_parsing_errors=True  # 自动处理解析错误
)

# 测试
result = agent_executor.invoke({
    "input": "我持有苹果10股、谷歌5股，告诉我组合总价值，以及最新的AI市场动态"
})
print(result["output"])

4.3 踩坑实录三：Agent 无限循环与 max_iterations

现象：Agent 一直在调用工具，日志里出现大量重复的 Thought/Action，几分钟后报 Agent stopped due to iteration limit or time limit。

原因：通常有两种情况：

  1. 工具返回的结果格式 Agent 看不懂，导致它认为任务没完成，持续尝试
  2. 任务描述模糊，Agent 不知道什么时候该停下来

解法：

# 1. 工具返回值要结构化，给 Agent 明确的成功/失败信号
@tool
def query_database(sql: str) -> str:
    """查询数据库"""
    try:
        result = db.execute(sql)
        if result:
            return f"查询成功，返回 {len(result)} 条数据：{result[:3]}..."  # 明确告知成功
        return "查询成功，无数据"  # 明确告知无数据
    except Exception as e:
        return f"查询失败：{str(e)}"  # 明确告知失败原因

# 2. 设置合理的 max_iterations 和 early_stopping_method
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,
    early_stopping_method="generate",  # 超过迭代次数后生成最终答案
    verbose=True
)

五、三者组合：构建一个完整的 AI 助手

真实项目中，Chain、Memory、Agent 往往需要配合使用。下面是一个结合三者的生产级示例：

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.memory import ConversationBufferWindowMemory
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import tool

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

# 定义工具
@tool
def search_knowledge_base(query: str) -> str:
    """搜索内部知识库"""
    # 实际项目接入向量数据库
    return f"知识库搜索结果：关于 '{query}' 的相关内容..."

@tool  
def create_jira_ticket(title: str, description: str, priority: str = "Medium") -> str:
    """创建 Jira 工单"""
    ticket_id = f"PROJ-{hash(title) % 10000}"
    return f"工单已创建：{ticket_id}，标题：{title}，优先级：{priority}"

tools = [search_knowledge_base, create_jira_ticket]

# 带 Memory 的 Agent Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是公司内部 IT 助手，帮助员工解决技术问题和创建工单。记住用户之前说过的内容。"),
    MessagesPlaceholder(variable_name="chat_history"),  # 注入历史对话
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# 共享 Memory
memory = ConversationBufferWindowMemory(
    k=6,
    return_messages=True,
    memory_key="chat_history",
    output_key="output"  # 重要：Agent 有多个输出键，要指定哪个存入 Memory
)

agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=False,
    max_iterations=8,
    handle_parsing_errors=True
)

# 多轮对话测试
print(agent_executor.invoke({"input": "我的 Jenkins 流水线一直报 OOM 错误"})["output"])
print(agent_executor.invoke({"input": "帮我创建一个工单，就是刚才说的那个问题"})["output"])
# 第二轮会记住"刚才的问题"是 Jenkins OOM，自动填充工单内容

六、性能与成本数据参考

基于我实际项目的测量数据（gpt-3.5-turbo，中文场景）：

场景平均延迟Token 消耗成本估算（$/千次）
简单 LLMChain1.2s~800 tokens~$0.12
SequentialChain（3步）3.8s~2400 tokens~$0.36
ConversationChain（含5轮历史）2.1s~1500 tokens~$0.22
Agent（平均3次工具调用）8.5s~3500 tokens~$0.52

Agent 的成本是普通 Chain 的4-5倍，如果不是必须让模型自主决策，优先用确定性的 Chain。

七、选型建议

根据我的实战经验，给出一个决策框架：

  • 流程固定，步骤明确 → 用 Chain（成本低、速度快、可预期）
  • 需要多轮对话 → 加 Memory（选合适类型控制 Token 消耗）
  • 需要动态决策，根据情况选择不同工具 → 用 Agent（慎用，成本高）
  • 生产环境，并发量大 → 避免用 Agent，能 Chain 化就 Chain 化

LangChain 这三个组件不是非此即彼的，灵活组合才是正道。

编辑此页
最近更新:
贡献者: xiaozhangge