本文手把手教你从零搭建一个企业级 AI Agent 中台，核心围绕 AI Agent 执行引擎、LLM 适配层和RAG 知识管线三大主题展开。我将用 30 天实战经验，拆解一个名为 DOSM（Digital Operations Smart Manager）的运维智能体中台架构。文章涵盖六层架构设计、ReAct 循环实现、Python + Java 异构集成方案，以及完整的五阶段开发路线图。无论你是想深入理解 Agent 底层原理，还是准备在面试中应对“Agent 执行引擎怎么设计”这类问题，本文都能给你可落地的答案。

读者收获：

- 掌握 AI Agent 中台的核心数据流与六层架构设计

- 学会用适配器模式统一管理 DeepSeek、Qwen、OpenAI 等多模型

- 理解 ReAct 循环的手写实现思路，不依赖 LangChain Agent

- 获得一套完整的 RAG 知识管线（文档解析 → 向量化 → 混合检索）实战方案

---

目录

1. 为什么我要手写一个 AI Agent 中台

2. 核心数据流：一次对话在系统里怎么跑

3. 六层架构逐层拆解

- 3.1 接入层：SSE vs WebSocket 选型

- 3.2 Agent 层：三种执行策略设计

- 3.3 记忆层：四级记忆体系

- 3.4 知识层：五步 RAG 管线

- 3.5 基础设施与可观测

4. Python + Java 异构架构实战

5. 五阶段开发路线图

6. 代码量统计与模块拆解

7. 总结与源码获取

---

1. 为什么我要手写一个 AI Agent 中台

先交代一下背景。我是做运维开发的，平时跟 CMDB、工单系统、告警中心打交道。今年上半年 AI Agent 这个概念特别火，我就想：能不能把 LLM 接进运维系统里，让它自动处理告警、查日志、甚至帮我写运维报告？

研究了市面上的方案之后，我发现一个问题。

Coze、Dify 这些平台很好用，拖拽就能搭 Agent。但它们更像给非技术人员用的工具，不是给开发者用的架构。你很难理解底层是怎么跑的——ReAct 循环到底怎么实现？工具调用的 Schema 怎么传给 LLM？SSE 流式输出有没有更好的方式？

最关键的是，面试官不会问你“怎么用 Dify 搭工作流”。他会问：“你的 Agent 执行引擎怎么设计的？”

所以我决定自己写一个。不是为了重复造轮子，是为了真的搞懂。

这个项目要解决什么问题

一句话：让运维从“被动应答”变成“主动理解、自主决策、持续记忆、多角色协同”。

翻译成人话就是——凌晨三点告警响了，不用把人叫起来，Agent 自己去看日志、定位问题、给出处理建议。搞不定的再升级给人。

我给它起了个名字叫 DOSM，Digital Operations Smart Manager。目前还处在早期阶段，但核心链路已经跑通了。

---

2. 核心数据流：一次对话在系统里怎么跑

先不管代码，想清楚数据流。用户发来一条消息之后，经过这些步骤：

用户消息 → API Gateway → AgentService
│
┌──────┼──────┐
▼      ▼      ▼
意图识别 记忆加载 知识检索
│      │      │
└──────┼──────┘
       ▼
Agent 执行引擎
(ReAct/Plan-Exec/Reflect)
       │
┌──────┼──────┐
▼      ▼      ▼
LLM调用 工具调用 结果合成
│      │      │
└──────┼──────┘
       ▼
SSE 流式返回给用户
       │
┌──────┴──────┐
▼             ▼
记忆更新      用户看到回复

7 个节点，每个节点后面都是一个独立的模块。这个流程我画了很多遍，每次画都能发现新问题——比如“工具调用结果太长把上下文撑爆了怎么办”、“LLM 一直要求调同一个工具怎么办”——这些后面会细讲。

---

3. 六层架构逐层拆解

把这个数据流拆成架构层，我分成了六层：

接入层 │ 钉钉/企微/飞书/Web/SSE
Agent层 │ 意图识别 + 多轮对话 + 任务编排 + 工具调用
       │ ReAct · Plan-Exec · Reflect
记忆层 │ 槽位(Slots) + 工作记忆 + 长期记忆 + 原子快照
知识层 │ 文档解析 → Embedding → BM25+向量 → Rerank
基础设施 │ Nacos + Redis + PostgreSQL + RocketMQ
可观测 │ Langfuse(LLM Trace) + Prometheus + Grafana

3.1 接入层：SSE vs WebSocket 选型

AI 在企业里不是孤立的东西。钉钉机器人要能调它，企业微信要能调它，Web 页面也要能调它。接入层的职责很简单：协议适配 → 鉴权 → 路由，不做业务逻辑。

选 SSE 而不是 WebSocket 的原因很简单：对话场景 99% 是服务端推送、客户端接收。不需要双向通信，SSE 更轻量，基于 HTTP，天然支持自动重连。

技术要点：SSE 流式输出时，每个 token 以 data: {content: "..."}\n\n 格式推送，前端用 EventSource 接收，天然支持断线重连。

3.2 Agent 层：三种执行策略设计

这是最核心的部分。一个请求进来之后：

1. 先做意图识别——用户想查知识？还是执行操作？还是在排查故障？

2. 加载会话记忆——这哥们刚才聊到哪了

3. 检索知识库——有没有相关的运维文档

4. 进执行引擎——决定怎么处理

执行引擎我设计了三种策略：

- ReAct：最常用的一种。Think（想一下该干嘛）→ Act（执行工具）→ Observe（看结果）→ 循环。适合大部分场景

- Plan-Exec：先让 LLM 出一个 N 步计划，再一步步执行。适合复杂任务，比如“帮我分析上周的告警趋势然后出个报告”

- Reflect：执行完了自己审一遍，不满意重来。适合对质量要求高的场景

目前 Phase 1 只实现了 ReAct。Plan-Exec 和 Reflect 在 Phase 3 做。

下面是 ReAct 循环的核心伪代码实现：

# ReAct 执行引擎核心逻辑
class ReActEngine:
    def __init__(self, llm, tools, max_iterations=10):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations

    async def execute(self, user_input, memory, knowledge):
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": user_input}
        ]
        # 注入记忆和知识
        messages = self._inject_context(messages, memory, knowledge)

        for step in range(self.max_iterations):
            # Step 1: Think - 让 LLM 决定下一步
            response = await self.llm.chat(messages)

            # Step 2: 解析 LLM 输出，判断是调工具还是直接回答
            action = self._parse_action(response)

            if action.type == "final_answer":
                return action.content

            # Step 3: Act - 执行工具
            tool_result = await self.tools.execute(
                action.tool_name,
                action.tool_args
            )

            # Step 4: Observe - 把结果放回上下文
            messages.append({
                "role": "assistant",
                "content": response
            })
            messages.append({
                "role": "tool",
                "tool_call_id": action.tool_name,
                "content": tool_result
            })

        return "抱歉，我无法在限定步数内完成这个任务。"

3.3 记忆层：四级记忆体系

这是我觉得最有意思、也最复杂的一层。

大部分 AI 项目的“记忆”就是一个聊天记录列表。但在运维场景里不够用。比如用户说“查一下 mysql-prod-01 的 CPU”，Agent 需要知道 mysql-prod-01 是一个资产 ID、CPU 是指标类型、时间范围默认最近 1 小时。这些叫“槽位”（Slots），是需要从对话里提取出来的结构化信息。

我设计了四级记忆：

- 槽位记忆：结构化的关键信息提取（资产 ID、集群名、时间范围、故障类型）

- 工作记忆：当前对话窗口内的事情，Token 预算管理

- 长期记忆：跨会话的经验沉淀，比如“上次这个集群出问题是因为连接池满了”

- 原子快照：整个对话上下文的 JSONB 存档，用于审计和故障复盘

前两级 Phase 1 实现了（Redis 做热缓存），后两级是 Phase 4 的事情。

记忆层实现要点：使用 Redis Hash 存储会话记忆，key 为 session:{session_id}:memory，field 为记忆层级（slot/working/long_term/snapshot），value 为 JSON 序列化后的记忆内容，TTL 设置 24 小时。

3.4 知识层：五步 RAG 管线

运维团队通常有大量文档——运维手册、故障案例、配置规范——散落在 Confluence、语雀、本地 Markdown 里。知识层就是把这些东西“喂”给 AI。

我设计的 RAG 管线分五步：

1. 文档解析：PDF（含表格）、DOCX、HTML、MD、TXT 五种格式，一个入口

2. 智能切分：段落优先 → 句子降级 → 字符硬切，保证 chunk 在 500 token 左右

3. 向量化：BGE / Text2vec / Qwen Embedding 三种后端可切换

4. 混合检索：BM25 关键词 + 向量语义 + RRF 融合 + Reranker 精排

5. 溯源问答：每个回答标注来源“《MySQL 运维手册》第 42 页”

前两步（解析 + 切分 + 向量化 + Milvus 入库）Phase 2a 已经实现了。混合检索和溯源问答 Phase 2b/2c 在做。

# 文档管线核心实现示例
from typing import List, Dict, Any
import hashlib

class DocumentPipeline:
    def __init__(self, embedder, vector_store):
        self.parsers = {
            ".pdf": PDFParser(),
            ".docx": DOCXParser(),
            ".html": HTMLParser(),
            ".md": MDParser(),
            ".txt": TXTParser()
        }
        self.splitter = SmartSplitter(chunk_size=500, chunk_overlap=50)
        self.embedder = embedder
        self.vector_store = vector_store

    async def process_document(self, file_path: str) -> Dict[str, Any]:
        # Step 1: 解析文档
        ext = self._get_extension(file_path)
        parser = self.parsers.get(ext)
        if not parser:
            raise ValueError(f"不支持的文档格式: {ext}")
        raw_text = await parser.parse(file_path)

        # Step 2: 智能切分
        chunks = self.splitter.split(raw_text)

        # Step 3: 向量化
        embeddings = await self.embedder.embed_documents(chunks)

        # Step 4: 入库（Milvus）
        documents = []
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            doc = {
                "id": hashlib.md5(chunk.encode()).hexdigest(),
                "text": chunk,
                "embedding": embedding,
                "metadata": {
                    "source": file_path,
                    "chunk_index": i,
                    "total_chunks": len(chunks)
                }
            }
            documents.append(doc)
        await self.vector_store.insert(documents)

        return {
            "status": "success",
            "total_chunks": len(chunks),
            "document_id": hashlib.md5(file_path.encode()).hexdigest()
        }

3.5 基础设施和可观测

基础设施选型没什么特别的：Redis 做热缓存、PostgreSQL 持久化、Milvus Standalone 做向量存储。Phase 4-5 接入 Java 业务层之后加上 Nacos 服务发现和 RocketMQ 消息队列。

可观测这块很重要但容易被忽略。LLM 调用链不可观测 = 生产事故没法排查。计划集成 Langfuse 做 LLM Trace（每次调用的 Prompt、Token、延迟），Prometheus + Grafana 做服务监控。

---

4. Python + Java 异构架构实战

很多人问我：为什么不统一用一种语言？

答案很简单：AI 生态是 Python 的。LangChain、Transformers、PyTorch、Sentence-Transformers——都是 Python 原生。你用 Java 调 LLM 不是不行，但每接入一个新模型都要自己写 HTTP Client + Parser，不值当。

但企业已有的 CMDB、工单、告警系统大概率是 Java 的。这些系统不需要重写，做好服务发现就行。

所以策略是：Phase 1-3 纯 Python，把 AI 能力跑通。Phase 4-5 通过 Nacos 接入 Java 业务层。先做核心，再补周边。

Java (SpringCloud) Python (FastAPI + LangChain)

业务服务层 AI 编排层

┌──────────────┐ ┌──────────────────────┐

│ CMDB/资产 │◄──Nacos──► │ AgentService │

│ 工单系统 │ 服务发现 │ ├─ LLM Gateway │

│ 告警中心 │ │ ├─ Agent Engine │

│ 用户/权限 │ │ ├─ RAG Pipeline │

│ 定时调度 │ │ ├─ Memory System │

└──────────────┘ │ └─ Tool Framework │

└──────────────────────┘

 异构架构核心设计：Python 侧暴露 RESTful API（FastAPI），Java 侧通过 Nacos 服务发现调用。Python 侧使用 `nacos-sdk-python` 注册服务，Java 侧通过 `@FeignClient` 声明式调用。关键数据格式统一使用 JSON Schema 约束。

---

5. 五阶段开发路线图

我没打算一口气做完。分五个里程碑，每个有明确的验收标准：

| 里程碑 | 阶段 | 演示什么 | 当前状态 |

|--------|------|---------|---------|

| M1: Hello Agent | Phase 1 | "今天几号？100+200=？"→ Agent 调工具 → 流式回答 | ✅ 完成 |

| M2: 知识问答 | Phase 2 | 导 20 份运维文档，问"MySQL 连接池满了怎么处理？"→ 溯源回答 | 🔶 2a 完成 |

| M3: 多策略 | Phase 3 | 同一任务 ReAct vs Plan-Exec 对比 | 📐 设计阶段 |

| M4: 长记忆 | Phase 4 | 50 轮对话后还能记住关键信息 | 📐 设计阶段 |

| M5: 故障自愈 | Phase 5 | 模拟告警→排查→定位→出报告 | 📐 设计阶段 |

M1 的验收标准我全过了，挑几个关键的：

- Docker Compose 一键启动，`/api/v1/health` 返回 healthy

- 改 `.env` 里 `LLM_PROVIDER=deepseek→qwen` 重启就切换，不用改代码

- "今天几号"自动调用时间工具，"帮我算 123*456"自动调计算器

- SSE 流式逐 token 推送

- 不带 API Key 返回 401

- LLM 挂了返回"AI 服务暂时不可用"，不暴露 API Key

---

6. 代码量统计与模块拆解

不算测试，纯 Python 大概 1500 行。拆一下：

- **LLM 适配层**：~250 行（Abstract Base + OpenAI Adapter + Factory + Retry）

- **工具框架**：~200 行（Registry + 3 个内置工具 + 安全计算器）

- **ReAct 执行引擎**：~200 行（循环 + 工具执行 + 截断 + 错误处理）

- **会话记忆**：~120 行（Redis CRUD + TTL + 列表查询）

- **文档管线**：~600 行（5 种解析器 + 切分器 + 3 种 Embedder + Milvus CRUD）

- **API 层**：~300 行（Chat/Session/Knowledge 路由 + 中间件 + 健康检查）

7个 API 端点，5 个 Docker 容器（agent + postgres + redis + milvus + etcd/minio）。

---