第 04 期 | 拆解工业智能体技术栈：Agent / MCP / RAG / 工具调用 / 记忆 🧠
摘要：一个可落地的工业智能体，绝不是一个模型那么简单，而是一套**编排框架（Orchestration）+ 工具协议（Protocol）+ 记忆机制（Memory）+ 检索增强（RAG）协同工作的系统工程。本文从底层逐层解构：首先厘清Agent 编排（以 LangGraph 为代表的有状态图编排）与工具调用（Function Calling）的实现细节，其次拆解MCP如何让智能体"即插即用"接入工业系统，再次澄清RAG 与 Agentic RAG的本质差异，最后给出记忆（Memory）**的分类与工程实现，并用一段可运行代码把五者串成完整闭环。
〇、引言：别把"技术栈"看成卖点，它是地基
一、Agent 编排：为什么需要"有状态图"而不是 if-else 🧩
工业任务不是单次问答，而是多步骤、可能有分支回退、需要记忆上下文的流程。学术与工程上通常用**图编排（Graph Orchestration）**来描述，例如 LangGraph 将"节点（Node）+ 边（Edge）+ 共享状态（State）"显式建模：
💡知识卡片：与线性脚本不同，有状态图允许节点间共享可变状态、支持条件分支与循环回退，这正是工业任务（如"先读数据→判断是否异常→异常则调用告警工具→记录并反馈"）所需要的控制结构。
二、工具调用（Function Calling）：智能体"动手"的机制 🛠️
LLM 本身只输出文本；要"动手"，必须把外部能力封装为**工具（Tool）**并让模型按结构化参数调用。其标准流程是：
用户/系统 →[意图]→ LLM(带工具schema) →[tool_call(名称,参数)]→ 执行器 →[tool_result]→ LLM汇总下面用 LangGraph 给出一个"查询炉况并告警"的最小有状态 Agent 示例：
# agent/bf_agent.py"""LangGraph 有状态编排 + 工具调用：高炉炉况查询与告警"""fromtypingimportAnnotated,TypedDict,Literalfromlangchain_openaiimportChatOpenAIfromlangchain.toolsimporttoolfromlanggraph.graphimportStateGraph,ENDfromlanggraph.graph.messageimportadd_messagesclassBFState(TypedDict):messages:Annotated[list,add_messages]# 共享对话状态(记忆载体)urgent:bool@tooldefquery_hot_blast_pressure()->float:"""查询当前热风压力(MPa)。"""# 真实场景由第13期 OT-IT 采集工具提供return0.38@tooldefissue_alert(level:Literal["warn","critical"],msg:str)->str:"""发送炉况告警。level: warn=警告, critical=紧急。"""# 真实场景对接短信/飞书/工控联动returnf"[{level.upper()}]{msg}"defdecide(state:BFState)->BFState:state["urgent"]="热风压力"instr(state["messages"][-1].content)returnstatedeffinalize(state:BFState)->BFState:p=query_hot_blast_pressure.invoke({})returnBFState(messages=[("assistant",f"当前热风压力{p}MPa")])defbuild_graph(llm):g=StateGraph(BFState)g.add_node("decide",decide)g.add_node("finalize",finalize)g.add_edge("decide","finalize")g.add_conditional_edges("decide",lambdas:"finalize"ifnots["urgent"]else"finalize")g.set_entry_point("decide")returng.compile()📌实战要点：工具必须"小而准、有明确 schema、能失败返回"（详见错误中间件）。否则一次
tool_call解析失败，就会让整个 loop 卡死。工业上建议对所有工具统一加wrap_tool_call错误兜底。
三、MCP：把"接口集成"变成"即插即用" 🔌
MCP（Model Context Protocol）本质是定义了模型客户端与工具/数据服务器之间的一个标准层。其二段式结构极大降低了"为每家新系统写一遍对接代码"的重复劳动：
⚠️提醒：MCP 解决的是"协议统一"，不等于"安全已就绪"。工业现场仍必须做权限白名单、审计日志、操作复核（第 27 期详述）。
四、RAG 与 Agentic RAG：从"查文档"到"带文献决策" 📚
经典 RAG 是"检索→拼进上下文→生成"的单程；而Agentic RAG让智能体自己决定"检索什么、要不要二次检索、如何引用"。这对高炉场景意义重大——炉况异常时，智能体应先检索同类事故案例与操作标准，再结合实时数据给出处置建议。
| 维度 | 经典 RAG | Agentic RAG |
|---|---|---|
| 检索触发 | 固定每次先检索 | 按需、多轮、迭代检索 |
| 决策依据 | 一次拼接 | 检索结果+实时数据融合 |
| 适用 | 知识问答 | 诊断、处置建议、复核 |
五、记忆：让智能体"不再是每次失忆的新员工" 🧠
工业智能体至少要具备三类记忆：
| 记忆类型 | 载体 | 作用 | 工程实现 |
|---|---|---|---|
| 短期/会话记忆 | checkpointer | 本轮对话不丢 | MemorySaver/SQLite |
| 长期/事实记忆 | 向量库/DB | 跨会话记住炉子个性 | 关键指标持久化 |
| 工具/过程记忆 | 日志/DB | 沉淀经验、可复盘 | 事件表+审计 |
下面给出短期记忆的关键实现（滑动窗口聚合消息）：
# agent/memory.pyfromtypingimportAnnotatedfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportAnyMessage MAX_MESSAGES=40# 默认保留最近20轮(40条)def_windowed(old,new):"""滑动窗口：只保留最近 MAX_MESSAGES 条，防止上下文无限膨胀"""returnadd_messages(old,new)[-MAX_MESSAGES:]classAgentState(dict):messages:Annotated[list[AnyMessage],_windowed]# type: ignore🎯要点：工业场景记忆要"可控可删"。涉及配方、工艺参数的长期记忆，务必做版本化与审计，不可让模型在长程对话后"悄悄改变"关键参数。
🎯 本期小结
- Agent 编排建议用有状态图（LangGraph 等），天然支持分支回退与共享状态；
- 工具调用是智能体"动手"的机制，务必"小而准 + 可失败 + 有 schema"；
- MCP让系统接入"即插即用"，但安全（白名单/审计）仍需另行加固；
- Agentic RAG优于经典 RAG，能让智能体"带文献+带实时数据"做处置决策；
- 记忆分三类（会话/事实/过程），工程上要"窗口化控住长度 + 版本化控住参数"。
🔧 本期工具箱
- 编排框架：LangGraph（本文示例），可与自研 Agent 结合；
- 记忆实现：
_windowed滑动窗口 + MemorySaver checkpointer（第 10 期也会用到）； - 延伸阅读：Anthropic《Building Effective Agents》、MCP 官方规范文档。
下一期预告：技术与架构都清楚了，剩最后一关——“思维”。第 05 期《从"AI助手"到"工业智能体"：能力跃迁与落地范式》，讲明为什么"会聊天"离"会干活"还差一个范式。
作者：高炉炼铁智能化技术研究者，专注钢铁冶金与人工智能 +交叉领域。
👍 如果觉得有帮助，请点赞、收藏、转发！
版权归作者所有，未经许可请勿抄袭，套用，商用(或其它具有利益性行为)。
🔔 关注专栏，不错过后续精彩内容
