第17章 LangChain实战：自定义Agent、工具链与记忆系统

LangChain实战：自定义Agent、工具链与记忆系统

上一章我们学习了LangChain的基础概念和基本用法。本章将深入LangChain的实战应用——自定义Agent开发、复杂工具链设计、记忆系统优化、生产级应用架构等。通过本章的学习，你将掌握用LangChain构建生产级智能体应用的能力。

一、自定义Agent开发

1.1 Agent的核心组件

一个LangChain Agent由以下核心组件组成：

• LLM：大语言模型，负责推理和决策
• Tools：工具列表，Agent可以调用的外部能力
• Prompt：提示词，定义Agent的角色、行为规则、输出格式
• Memory：记忆系统，存储对话历史和上下文
• Output Parser：输出解析器，解析Agent的思考和行动
• Agent Executor：执行器，管理Agent的执行循环（思考→行动→观察→再思考）

1.2 自定义Agent的步骤

除了使用LangChain内置的Agent（如create_tool_calling_agent），你也可以自定义Agent，实现更灵活的逻辑。

from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.output_parsers import BaseOutputParser
from typing import Union
import json
import re

# 第一步：自定义输出解析器
class CustomAgentOutputParser(BaseOutputParser):
    """解析Agent的输出，判断是行动还是最终回答"""
    
    def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
        # 检查是否包含最终回答
        if "Final Answer:" in text:
            final_answer = text.split("Final Answer:")[-1].strip()
            return AgentFinish(
                return_values={"output": final_answer},
                log=text
            )
        
        # 解析工具调用（支持JSON格式）
        try:
            # 尝试解析JSON格式的工具调用
            json_match = re.search(r'\{[^{}]*\}', text)
            if json_match:
                action = json.loads(json_match.group())
                return AgentAction(
                    tool=action.get("tool", ""),
                    tool_input=action.get("tool_input", {}),
                    log=text
                )
        except:
            pass
        
        # 默认返回思考中
        return AgentAction(
            tool="continue",
            tool_input={},
            log=text
        )

# 第二步：自定义提示词
CUSTOM_AGENT_PROMPT = """你是一位专业的制造业设备故障诊断专家。
你的任务是根据用户描述的故障现象，调用工具查询相关信息，分析故障原因，给出维修方案。

可用工具：
{tools}

使用格式：
Question: 用户的问题
Thought: 你的思考过程
Action: 要调用的工具名称（JSON格式：{{"tool": "工具名", "tool_input": {{参数}}}}）
Observation: 工具返回的结果
...（重复Thought/Action/Observation直到得出结论）
Thought: 我现在有足够的信息给出最终答案
Final Answer: 最终的诊断报告

开始！
Question: {input}
Thought: {agent_scratchpad}"""

# 第三步：构建自定义Agent
from langchain.agents import AgentExecutor
from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(CUSTOM_AGENT_PROMPT)
parser = CustomAgentOutputParser()

# 构建LLM链
llm_chain = prompt | llm | parser

# 创建Agent执行器
agent_executor = AgentExecutor(
    agent=llm_chain,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=10,  # 最大迭代次数，防止无限循环
    early_stopping_method="generate",  # 提前停止策略
    handle_parsing_errors=True,  # 处理解析错误
)

# 调用
result = agent_executor.invoke({"input": "3号空压机异响，温度95度"})
print(result["output"])

1.3 Agent的高级配置

• max_iterations：最大迭代次数，防止Agent无限循环。一般设为5-15
• early_stopping_method：提前停止策略。"force"强制停止，"generate"让Agent生成最终答案
• handle_parsing_errors：处理解析错误。可以设为True（自动处理）或自定义错误处理函数
• return_intermediate_steps：是否返回中间步骤（思考过程、工具调用），用于调试和分析
• max_execution_time：最大执行时间（秒），超时自动停止

二、复杂工具链设计

2.1 工具的高级用法

工具的结构化输入：

from langchain_core.tools import tool
from pydantic import BaseModel, Field

# 定义工具输入结构
class QueryEquipmentInput(BaseModel):
    equipment_id: str = Field(description="设备编号，如'COMP-003'")
    include_history: bool = Field(default=False, description="是否包含历史维修记录")

# 使用结构化输入定义工具
@tool(args_schema=QueryEquipmentInput)
def query_equipment_info(equipment_id: str, include_history: bool = False) -> dict:
    """查询设备档案信息，包括设备型号、参数、安装时间、维护记录。
    当用户询问设备信息、历史故障、维护记录时使用。"""
    # 实际实现：调用企业设备管理系统API
    return {
        "equipment_id": equipment_id,
        "model": "螺杆式空压机GA75",
        "install_date": "2020-03-15",
        "power": "75kW",
        "history": [...] if include_history else []
    }

工具的错误处理：

@tool
def create_maintenance_order(equipment_id: str, fault_description: str, priority: str = "normal") -> dict:
    """创建维修工单。参数：equipment_id（设备编号）、fault_description（故障描述）、priority（优先级：low/normal/high/urgent）。
    当用户要求创建工单、安排维修时使用。"""
    try:
        # 调用工单系统API
        response = requests.post(
            "https://api.example.com/maintenance-orders",
            json={
                "equipment_id": equipment_id,
                "fault_description": fault_description,
                "priority": priority
            },
            timeout=10
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.Timeout:
        return {"success": False, "error": "工单系统超时，请稍后重试或联系系统管理员"}
    except requests.exceptions.RequestException as e:
        return {"success": False, "error": f"创建工单失败：{str(e)}"}
    except Exception as e:
        return {"success": False, "error": f"未知错误：{str(e)}"}

2.2 工具链的编排模式

模式一：顺序工具链（Sequential）

工具按固定顺序执行，前一个工具的输出作为后一个的输入。适合流程固定的任务。

from langchain.chains import SequentialChain

# 定义各个步骤
step1 = query_equipment_info  # 查询设备档案
step2 = retrieve_maintenance_manual  # 检索维修手册
step3 = analyze_fault_cause  # 分析故障原因（LLM节点）
step4 = generate_repair_plan  # 生成维修方案（LLM节点）
step5 = create_maintenance_order  # 创建工单

# 构建顺序链
chain = SequentialChain(
    chains=[step1, step2, step3, step4, step5],
    input_variables=["equipment_id", "fault_description"],
    output_variables=["diagnosis_report", "order_id"]
)

模式二：路由工具链（Router）

根据条件选择不同的工具或子链。适合需要分支处理的任务。

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

# 定义不同场景的处理链
chains = {
    "equipment_fault": equipment_fault_chain,  # 设备故障处理
    "process_optimization": process_optimization_chain,  # 工艺优化
    "quality_issue": quality_issue_chain,  # 质量问题处理
    "energy_management": energy_management_chain,  # 能源管理
}

# 构建路由链
router_chain = LLMRouterChain.from_llm(llm, router_prompt)
chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=chains,
    default_chain=default_chain,
    verbose=True
)

模式三：并行工具链（Parallel）

多个工具同时执行，最后汇总结果。适合没有依赖关系的任务。

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

# 并行执行多个检索
parallel_retrieval = RunnableParallel({
    "equipment_info": query_equipment_info,
    "maintenance_manual": retrieve_maintenance_manual,
    "fault_history": query_fault_history,
    "similar_cases": search_similar_cases
})

# 汇总后交给LLM分析
analysis_chain = parallel_retrieval | analysis_prompt | llm | output_parser

2.3 工具的最佳实践

• 工具数量控制：一个Agent的工具数量控制在5-15个。太多会导致选择困难，调用准确率下降
• 工具描述清晰：每个工具的名称和描述要准确，说明"做什么、什么时候用、返回什么"。Agent是根据描述来判断是否调用的
• 参数结构合理：用Pydantic定义参数结构，有类型、有描述、有默认值。参数不要太复杂，嵌套不超过2层
• 错误处理完善：每个工具都要有完善的错误处理，返回清晰的错误信息，让Agent知道哪里错了、该怎么修正
• 结果结构化：工具返回JSON格式的结构化数据，不要返回纯文本。结构化数据更容易被Agent理解和使用
• 工具分组：工具很多时，按功能分组（如查询类、操作类、分析类），Agent可以先选组再选工具
• 动态加载：根据用户问题动态加载相关工具，而不是所有工具都加载。可以用一个"工具路由器"先判断需要哪类工具

三、记忆系统优化

3.1 记忆系统的设计

生产级应用的记忆系统需要考虑以下几个方面：

• 短期记忆：当前对话的上下文，存在内存或缓存中
• 长期记忆：用户的历史对话、偏好、画像，存在数据库或向量数据库中
• 知识记忆：企业知识库、文档、案例，存在向量数据库中（即RAG）
• 经验记忆：Agent的执行经验、成功/失败案例，用于持续优化

3.2 对话记忆的优化

滑动窗口记忆：

from langchain.memory import ConversationBufferWindowMemory

# 只保留最近5轮对话
memory = ConversationBufferWindowMemory(k=5, return_messages=True)

摘要记忆：

from langchain.memory import ConversationSummaryMemory

# 用大模型总结对话历史，节省上下文
memory = ConversationSummaryMemory(
    llm=llm,
    return_messages=True,
    buffer="用户是一位工厂维修工，主要咨询设备故障问题。"
)

实体记忆：

from langchain.memory import ConversationEntityMemory
from langchain.memory.prompt import ENTITY_MEMORY_CONVERSATION_TEMPLATE

# 提取对话中的实体（人名、设备名、地点等），单独存储
memory = ConversationEntityMemory(llm=llm)

# 使用时可以查询特定实体的信息
memory.load_memory_variables({"input": "3号空压机上次维修是什么时候？"})

知识图谱记忆：

from langchain.memory import ConversationKGMemory
from langchain_community.graphs import Neo4jGraph

# 用知识图谱存储对话中的实体和关系
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="password"
)
memory = ConversationKGMemory(llm=llm, graph=graph)

3.3 长期记忆的实现

对于需要长期保存的用户记忆（如用户偏好、历史对话、用户画像），可以用向量数据库存储：

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from datetime import datetime

class LongTermMemory:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = Chroma(
            collection_name=f"user_memory_{user_id}",
            embedding_function=self.embeddings,
            persist_directory="./user_memory"
        )
    
    def add_memory(self, content: str, metadata: dict = None):
        """添加一条记忆"""
        doc = Document(
            page_content=content,
            metadata={
                "user_id": self.user_id,
                "timestamp": datetime.now().isoformat(),
                **(metadata or {})
            }
        )
        self.vectorstore.add_documents([doc])
    
    def retrieve_memory(self, query: str, k: int = 5) -> list:
        """检索相关记忆"""
        docs = self.vectorstore.similarity_search(query, k=k)
        return [doc.page_content for doc in docs]
    
    def get_user_profile(self) -> dict:
        """获取用户画像（从记忆中提取关键信息）"""
        # 可以用LLM总结用户的偏好、习惯、历史等
        pass

3.4 记忆系统的最佳实践

• 分层记忆：短期记忆（滑动窗口）+ 中期记忆（摘要）+ 长期记忆（向量数据库），不同层级存储不同时效的信息
• 记忆筛选：不是所有对话都要存入长期记忆。只保存有价值的信息（用户偏好、重要决策、关键事实），过滤掉无意义的闲聊
• 记忆更新：用户的偏好可能变化，要支持记忆的更新和覆盖。用最新的信息替换过时的信息
• 记忆遗忘：长期记忆也不是永久保存。可以设置记忆的有效期，过期的记忆自动归档或删除，避免记忆库膨胀
• 隐私保护：用户记忆包含敏感信息，要做好加密和权限控制。用户有权查看、导出、删除自己的记忆
• 记忆可解释：Agent使用记忆时，要能说明"这个信息来自哪里"，增加可信度，也方便纠错

四、生产级应用架构

4.1 应用架构设计

一个生产级的LangChain应用通常包含以下层次：

• 接入层：API接口、Web界面、移动端、企业微信/钉钉/飞书集成
• 编排层：LangChain链/Agent编排、工作流引擎、任务调度
• 能力层：大模型调用、RAG检索、工具调用、记忆管理
• 数据层：向量数据库、关系数据库、缓存、对象存储
• 基础设施：容器化部署、负载均衡、监控告警、日志追踪

4.2 关键技术要点

流式输出：

# 流式输出，逐token返回给前端
for chunk in chain.stream({"input": user_query}):
    yield chunk  # 通过SSE或WebSocket推送给前端

异步处理：

# 异步调用，提高并发性能
result = await chain.ainvoke({"input": user_query})

# 异步批量处理
results = await chain.abatch([
    {"input": "query1"},
    {"input": "query2"},
    {"input": "query3"}
])

重试和降级：

from langchain_core.runnables import RunnableRetry
from tenacity import retry, stop_after_attempt, wait_exponential

# 配置重试策略
retry_chain = chain.with_retry(
    stop_after_attempt=3,  # 最多重试3次
    wait_exponential(multiplier=1, min=4, max=10)  # 指数退避
)

# 降级策略：主模型失败时用备用模型
try:
    result = primary_chain.invoke(input)
except Exception as e:
    logger.warning(f"主模型失败，切换备用模型：{e}")
    result = fallback_chain.invoke(input)

缓存：

from langchain.cache import InMemoryCache, RedisCache
import langchain

# 内存缓存（适合单实例）
langchain.llm_cache = InMemoryCache()

# Redis缓存（适合多实例、分布式）
langchain.llm_cache = RedisCache(redis_kwargs={"host": "localhost", "port": 6379})

# 缓存后，相同的输入不会重复调用LLM，直接返回缓存结果

可观测性：

# 使用LangSmith追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "你的LangSmith API Key"
os.environ["LANGCHAIN_PROJECT"] = "equipment-diagnosis-agent"

# 自定义回调
from langchain_core.callbacks import BaseCallbackHandler

class CustomCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        logger.info(f"LLM调用开始，输入长度：{len(prompts[0])}")
    
    def on_llm_end(self, response, **kwargs):
        logger.info(f"LLM调用结束，token消耗：{response.llm_output.get('token_usage')}")
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        logger.info(f"工具调用开始：{serialized.get('name')}，输入：{input_str}")
    
    def on_tool_end(self, output, **kwargs):
        logger.info(f"工具调用结束，输出长度：{len(str(output))}")

4.3 部署方案

• LangServe：LangChain官方的部署工具，一行代码把链部署为REST API，支持流式输出、批量处理、异步调用
• FastAPI：用FastAPI自己封装API，更灵活，可以自定义中间件、认证、限流等
• Docker容器化：把应用打包成Docker镜像，方便部署和扩展
• Kubernetes：大规模生产环境用K8s部署，支持自动扩缩容、滚动更新、负载均衡
• Serverless：用AWS Lambda、阿里云函数计算等Serverless平台，按需付费，免运维

4.4 性能优化

• 模型选择：简单任务用小模型（如GPT-3.5、DeepSeek-V2），复杂任务用大模型。不要所有任务都用最贵的模型
• 提示词优化：精简提示词，去掉不必要的内容，减少token消耗
• 缓存：对重复查询缓存结果，减少LLM调用
• 批处理：批量处理多个请求，提高吞吐量
• 异步：IO密集型操作用异步，提高并发性能
• 检索优化：减少检索结果数量，只传最相关的3-5条，减少上下文长度
• 流式输出：用流式输出提升用户体验，用户不需要等完整结果
• 预加载：提前加载模型、知识库、常用数据，减少首次响应延迟

4.5 成本控制

• 监控用量：记录每次调用的token消耗、模型类型、工具调用次数，定期统计成本
• 设置预算：设置日/月预算上限，超预算自动告警或降级
• 模型路由：根据任务复杂度自动选择模型，简单任务用便宜的模型
• 缓存策略：对常见问题缓存回答，减少LLM调用
• 限制使用：对免费用户设置调用次数限制，引导付费
• 优化提示词：精简提示词，减少每次调用的token数
• 批量处理：批量处理减少API调用 overhead

五、实战：构建生产级设备故障诊断Agent

下一章，我们将学习Agent Harness和新一代智能体开发框架——2025年智能体开发的最新趋势和技术。

📝 本章练习（5题）

答案与解析请在 互动学习看板 中作答查看。

📚 AI智能体实战训练营 课程目录（33 章）

1. 第1章 什么是AI智能体：从聊天机器人到自主决策系统
2. 第2章 智能体的四大核心能力：感知·思考·行动·记忆
3. 第3章 范式转移：智能体与传统软件的本质区别
4. 第4章 2025年主流智能体平台全景对比与选型
5. 第5章 智能体应用场景地图：从个人效率到企业级落地
6. 第6章 编排思维：如何把大模型变成可执行的业务流程
7. 第7章 Prompt工程实战：写出让智能体精准执行的指令
8. 第8章 工作流编排入门：节点、连线、条件分支与循环
9. 第9章 知识库RAG实战：让智能体拥有你的专业数据
10. 第10章 工具调用实战：搜索·计算·API·数据库全掌握
11. 第11章 多智能体协作：角色分工、任务分发与结果汇总
12. 第12章 Coze/扣子入门：30分钟搭建你的第一个智能体
13. 第13章 Coze进阶：插件开发、工作流深度编排与知识库优化
14. 第14章 Dify入门：开源可私有化部署的智能体平台搭建
15. 第15章 Dify进阶：RAG管道优化、Agent模式与工作流编排
16. 第16章 LangChain入门：用代码构建可定制的智能体应用
17. 第17章 LangChain实战：自定义Agent、工具链与记忆系统
18. 第18章 Agent Harness与新一代框架：2025年智能体开发趋势
19. 第19章 平台选型决策树：不同业务场景下的最优平台选择
20. 第20章 任务拆解方法论：把复杂问题变成智能体可执行的步骤
21. 第21章 提示词编排艺术：System·User·Function的三层协同
22. 第22章 记忆系统设计：短期记忆、长期记忆与向量数据库选型
23. 第23章 错误处理与容错：智能体执行失败时的兜底策略
24. 第24章 评估与优化体系：如何量化衡量智能体的效果并持续改进
25. 第25章 安全与合规：智能体应用的数据安全、隐私保护与风险防控
26. 第26章 实战项目一：企业知识库智能问答助手（全流程实操）
27. 第27章 实战项目二：制造业设备故障诊断智能体
28. 第28章 实战项目三：供应链采购智能分析与比价助手
29. 第29章 实战项目四：多智能体协同的项目管理办公室（PMO）
30. 第30章 实战项目五：智能体与MES/ERP系统的深度集成
31. 第31章 智能体落地路线图：从POC验证到规模化应用的完整路径
32. 第32章 AI智能体工程师/产品经理的能力模型与成长路径
33. 第33章 作品集打造与面试准备：如何拿到AI智能体岗位Offer