链式调用与记忆系统

课程简介

Chain 组合、顺序链与对话记忆管理。

🎬 本课程视频：LangChain for LLM App Dev — LangChain 应用开发

---

链式调用与记忆系统

一、从单次调用到多步流程

单个 LLM 调用只能完成一个回合的处理。但在真实应用中，大部分任务需要多步骤处理：先理解输入、再检索信息、然后推理分析、最后生成输出。链式调用模式让这些步骤可以被结构化的组织和串联。

二、Chain 系统详解

2.1 顺序链（SequentialChain）

顺序链是最基础的链式模式：前一步的输出作为后一步的输入，形成一个流水线。

from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate

# 第一步：分析
analysis_prompt = PromptTemplate.from_template(
    "分析以下产品描述，列出 3 个核心卖点：
{product_description}"
)
analysis_chain = LLMChain(
    llm=llm,
    prompt=analysis_prompt,
    output_key="selling_points"
)

# 第二步：翻译
translation_prompt = PromptTemplate.from_template(
    "将以下卖点翻译成英文：
{selling_points}"
)
translation_chain = LLMChain(
    llm=llm,
    prompt=translation_prompt,
    output_key="english_selling_points"
)

# 第三步：润色
polish_prompt = PromptTemplate.from_template(
    "将以下英文卖点润色得更吸引人：
{english_selling_points}"
)
polish_chain = LLMChain(
    llm=llm,
    prompt=polish_prompt,
    output_key="polished_points"
)

# 组合为顺序链
full_chain = SequentialChain(
    chains=[analysis_chain, translation_chain, polish_chain],
    input_variables=["product_description"],
    output_variables=["polished_points"],
    verbose=True
)

result = full_chain.invoke({"product_description": "一款AI驱动的智能笔记应用"})
print(result["polished_points"])

输出键管理：每个子链的 output_key 定义了其输出在状态字典中的名称。后续链可以从状态字典中读取这些值作为输入。

2.2 路由链（RouterChain）

路由链根据输入的内容，选择不同的下游处理路径。

from langchain.chains.router import LLMRouterChain
from langchain.chains.router.llm_router import RouterOutputParser

# 定义不同领域的处理链
tech_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate.from_template(
        "你是一位技术专家。问题：{input}
请用技术术语回答。"
    )
)
business_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate.from_template(
        "你是一位商业分析师。问题：{input}
请从商业角度分析。"
    )
)
general_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate.from_template(
        "问题：{input}
请回答。"
    )
)

# 定义路由规则
router_prompt = PromptTemplate.from_template(
    '''根据用户问题，选择最合适的处理方向：

    问题：{input}

    可选方向：
    - tech: 技术类问题
    - business: 商业类问题
    - general: 其他问题

    只返回方向名称。'''
)

# 路由链根据输入选择子链
chain_map = {
    "tech": tech_chain,
    "business": business_chain,
    "general": general_chain
}

2.3 并行链（ParallelChain）

对于需要从多个角度同时分析同一个输入的场景，并行链让多个独立的链同时执行。

from langchain.chains import ConcurrentChain  # 或使用 LangGraph

# 多角度分析
def analyze_from_all_angles(question):
    '''从多个角度并行分析'''
    chains = {
        "technical": tech_chain,
        "business": business_chain,
        "ethical": ethical_chain
    }

    import concurrent.futures
    results = {}

    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = {
            executor.submit(chain.invoke, {"input": question}): name
            for name, chain in chains.items()
        }

        for future in concurrent.futures.as_completed(futures):
            name = futures[future]
            results[name] = future.result()

    return results

三、对话记忆（Memory）

3.1 为什么需要记忆？

LLM 本身是无状态的——每次调用都是独立的，不记得之前的对话。但大多数应用需要跨回合的上下文：客服系统需要知道用户之前说了什么，写作助手需要知道之前的修改历史。

3.2 记忆类型

ConversationBufferMemory——完整记录对话历史。

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory()
memory.chat_memory.add_user_message("你好，我叫张三")
memory.chat_memory.add_ai_message("你好张三，有什么可以帮助你的？")
memory.chat_memory.add_user_message("我刚才说了我叫什么名字？")

# 获取历史
history = memory.load_memory_variables({})
print(history["history"])

ConversationSummaryMemory——用摘要压缩对话历史。

from langchain.memory import ConversationSummaryMemory

summary_memory = ConversationSummaryMemory(
    llm=llm,  # 需要 LLM 来生成摘要
    max_token_limit=200  # 摘要的最大 Token 数
)

# 自动将长对话压缩为摘要
summary_memory.save_context(
    {"input": "我介绍了我的项目背景..."},
    {"output": "AI回复了一大段分析..."}
)

VectorStoreRetrieverMemory——用向量检索相关历史。

from langchain.memory import VectorStoreRetrieverMemory
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

# 初始化向量存储
vector_store = Chroma(embedding_function=OpenAIEmbeddings())

# 创建记忆
vector_memory = VectorStoreRetrieverMemory(
    retriever=vector_store.as_retriever(search_kwargs={"k": 3}),
    memory_key="relevant_history"
)

# 只检索与当前查询最相关的历史对话
relevant = vector_memory.load_memory_variables(
    {"input": "我之前问过关于价格的问题"}
)

3.3 Token 管理

Token 消耗是使用记忆系统时必须考虑的问题。

from langchain.memory import ConversationTokenBufferFixture

# 限制 token 数量的记忆
token_memory = ConversationTokenBufferMemory(
    llm=llm,
    max_token_limit=1000,  # 超过 1000 token 时截断
    memory_key="chat_history"
)

# 或者：先保留完整对话，超过阈值后转为摘要
from langchain.memory import ConversationSummaryBufferMemory

summary_buffer_memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,  # 超过此阈值时压缩为摘要
    memory_key="chat_history"
)

3.4 将记忆注入链

from langchain.chains import ConversationChain

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

conversation.invoke("你好")
# 回复：你好！有什么可以帮助你的？
conversation.invoke("我叫张三")
# 回复：你好张三！很高兴认识你。
conversation.invoke("我叫什么名字？")
# 回复：你刚才说你的名字是张三。

四、LangGraph：更灵活的图编排

对于复杂的、非线性的流程，LangChain 推出了 LangGraph——一个基于图的工作流引擎。

from langgraph.graph import StateGraph

# 定义状态
class AgentState(TypedDict):
    messages: list
    next_step: str

# 定义节点
def analyze(state: AgentState):
    # 分析输入
    return {"next_step": "search"}

def search(state: AgentState):
    # 搜索信息
    return {"next_step": "generate"}

def generate(state: AgentState):
    # 生成回答
    return {"messages": [...]}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("analyze", analyze)
graph.add_node("search", search)
graph.add_node("generate", generate)

graph.add_edge("analyze", "search")
graph.add_conditional_edges("search", decide_next_step)
graph.add_edge("generate", END)

app = graph.compile()

五、最佳实践

1. 选择合适的记忆类型：根据对话长度和重要性选择。短对话用 Buffer，长对话用 Summary，需要精确回忆用 VectorStore
2. 设置 Token 预算：为记忆分配固定的 Token 预算，防止历史对话占用太多上下文
3. 定期清理：会话结束后清理记忆，防止内存泄漏
4. 混合记忆策略：最近的对话用 Buffer 精确保存，较远的历史用 Summary 压缩

六、总结

链式调用将单步 LLM 调用升级为多步骤、可组合的工作流。顺序链适合流水线任务，路由链适合条件分支任务，并行链适合多角度分析任务。对话记忆系统为模型提供了跨回合的上下文感知能力，不同类型的记忆适应不同的使用场景。Token 管理策略在记忆系统的成本和收益之间取得平衡。

七、Token 管理策略

7.1 为什么需要管理 Token？

LLM 有上下文窗口限制（如 GPT-4 的 8K\/32K tokens）。当对话历史过长时，需要策略性地管理 Token。

7.2 Token 管理方案对比

方案	实现	优势	劣势
截断	删除最早的历史	简单	丢失信息
摘要	压缩历史为摘要	信息密度高	有损压缩
滑动窗口	只保留最近 N 轮	控制精确	丢失早期信息
向量存储	历史向量化后检索	无损	实现复杂

7.3 智能摘要记忆

from langchain.memory import ConversationSummaryMemory

summary_memory = ConversationSummaryMemory(
    llm=llm,
    max_token_limit=2000  # 摘要最大 token 数
)

八、总结

关键要点回顾：
- SequentialChain 和 LLMChain 的组合使用
- RouterChain 实现条件路由
- 四种记忆类型：Buffer、Window、Summary、Vector
- Token 管理策略：截断、摘要、滑动窗口
- 链和记忆是构建多步对话应用的基石

七、Token 管理策略

7.1 为什么需要管理 Token？

LLM 有上下文窗口限制（如 GPT-4 的 8K\/32K tokens）。当对话历史过长时，需要策略性地管理 Token。

7.2 Token 管理方案对比

方案	实现	优势	劣势
截断	删除最早的历史	简单	丢失信息
摘要	压缩历史为摘要	信息密度高	有损压缩
滑动窗口	只保留最近 N 轮	控制精确	丢失早期信息
向量存储	历史向量化后检索	无损	实现复杂

7.3 智能摘要记忆

from langchain.memory import ConversationSummaryMemory

summary_memory = ConversationSummaryMemory(
    llm=llm,
    max_token_limit=2000  # 摘要最大 token 数
)

八、总结

关键要点回顾：
- SequentialChain 和 LLMChain 的组合使用
- RouterChain 实现条件路由
- 四种记忆类型：Buffer、Window、Summary、Vector
- Token 管理策略：截断、摘要、滑动窗口
- 链和记忆是构建多步对话应用的基石

七、Token 管理策略

7.1 为什么需要管理 Token？

LLM 有上下文窗口限制（如 GPT-4 的 8K\/32K tokens）。当对话历史过长时，需要策略性地管理 Token。

7.2 Token 管理方案对比

方案	实现	优势	劣势
截断	删除最早的历史	简单	丢失信息
摘要	压缩历史为摘要	信息密度高	有损压缩
滑动窗口	只保留最近 N 轮	控制精确	丢失早期信息
向量存储	历史向量化后检索	无损	实现复杂

7.3 智能摘要记忆

from langchain.memory import ConversationSummaryMemory

summary_memory = ConversationSummaryMemory(
    llm=llm,
    max_token_limit=2000  # 摘要最大 token 数
)

八、总结

关键要点回顾：
- SequentialChain 和 LLMChain 的组合使用
- RouterChain 实现条件路由
- 四种记忆类型：Buffer、Window、Summary、Vector
- Token 管理策略：截断、摘要、滑动窗口
- 链和记忆是构建多步对话应用的基石

延伸阅读

• 📺 B 站播放列表：LangChain for LLM App Dev — LangChain 应用开发
• 📚 更多学习资源，请访问 deeplearning.ai 官网