如何从「单个智能助手/单条工作流」升级到
多智能体协同的企业 AI 中台设计。

---

在前六篇里，我们已经把企业智能助手从 0 做到了「能跑生产」：

1. ​（一）​ DeepSeek + 本地向量检索：企业知识库 RAG 问答
2. ​（二）​ 接入钉钉 / 企业微信：在 IM 里 @ 机器人就能用
3. ​（三）​ 多轮对话 + 工具调用：从 FAQ 到企业助手
4. ​（四）​ 反馈 + 日志闭环：越用越聪明
5. ​（五）​ LangChain / LlamaIndex：企业级智能工作流编排
6. ​（六）​ 接入监控与告警体系：可观测、可运维

但如果你在企业里真正铺开使用，很快会遇到几个现实问题：

• HR 想要自己的「人事助手」，客服要「客服助手」，运维要「故障诊断助手」，采购要「采购助手」……
• 各业务线都在接入自己的知识库、工具、流程，越来越像 N 个“烟囱式小系统”
• 「采购」流程要查 HR 政策、「客服」要访问订单、物流、财务系统，跨部门协同变得复杂
• 模型配额、算力、向量库、监控、权限，全都分散管理，效率和成本都很差

站在这一步往前看，你真正需要的已经不是更多“单点智能体”，而是：

一个 多智能体协同的企业 AI 中台：
统一支撑各个业务线的 AI 应用，又能让不同智能体之间顺畅协作。

这一篇，我们就从架构和代码两个层面，带你搭出这么一套 “AI 中台底座”。

---

一、从“一个助手”到“多智能体”：到底多在哪？

1.1 单智能体的边界

前几篇我们做的是「一个综合型助手」：

• 会查知识库（RAG）
• 会调一些工具（查订单、查工单、查 HR）
• 会多轮对话、澄清信息

它本质是一个“大而全的超级 Agent”，适合起步，但在企业里会遇到几个典型问题：

1. 专业性不够
   • HR 问题、法律问题、财务问题，全都一个 Agent 来答，很难在每个领域都做精
2. 扩展性差
   • 新增一个「风控助手」，要在一个超级 Prompt/代码中继续堆逻辑
3. 协作困难
   • 很多任务天然适合拆分成多个角色：
     需求分析 → 方案设计 → 审批决策 → 操作执行 → 总结汇报

1.2 多智能体模式带来的改变

在多智能体模式下，我们把“大而全”的助手拆成若干专职 Agent：

• HR Agent：只负责 HR 政策、薪酬、休假相关的问题和操作
• 采购 Agent：负责采购需求理解、下单、跟单
• 财务 Agent：负责报销、付款、预算控制
• 法务 Agent：负责合同条款风险扫描
• 运维 Agent：负责告警分析、故障诊断、变更建议

并让它们在统一中台内通过标准协议协作完成任务，例如：

员工：
「帮我采购一台 1.5 万的测试服务器，看下审批链路合不合规，并生成合同草案。」

可能涉及的协作链：

• 需求 → 采购 Agent 拆解 + 录入采购系统
• 审批链路是否合规 → HR/Finance/Policy Agent 查制度
• 合同草案 → 法务 Agent 基于模版 + 条款库生成草案
• 最终结果汇总 → 一个「Coordinator Agent」给员工汇总说明

这就是「多智能体协同」的真正价值。

---

二、企业 AI 中台：四层参考架构

结合前几篇的工作流与监控能力，可以抽象出一个清晰的「AI 中台」四层模型：

┌────────────────────────────────────────────┐
│               应用访问层（入口）           │
│  Web / App / 钉钉 / 企微 / API Gateway    │
└─────────────────────┬──────────────────────┘
                      │
┌─────────────────────▼──────────────────────┐
│            编排与协同层（中台大脑）        │
│  - 统一调度器（Orchestrator）             │
│  - 会话与任务状态管理                     │
│  - 多智能体路由与协作策略                 │
│  - 审计 & 日志 & 监控                     │
└─────────────────────┬──────────────────────┘
                      │
┌─────────────────────▼──────────────────────┐
│             智能体与能力层（AI 能力池）    │
│  - 领域 Agent：HR / 采购 / 客服 / 运维等   │
│  - 通用 Agent：RAG 文档助手 / 总结助手     │
│  - 工具 & 工作流：LangChain / LlamaIndex   │
└─────────────────────┬──────────────────────┘
                      │
┌─────────────────────▼──────────────────────┐
│             基础资源层（Infra & 数据）     │
│  - DeepSeek 等 LLM 服务                    │
│  - 向量库 / 文档索引（LlamaIndex 等）      │
│  - 业务系统 API（采购、HR、CRM、ITSM…）    │
│  - Redis / DB / MQ / 监控 / 日志            │
└────────────────────────────────────────────┘

中台的核心：“编排与协同层” + “智能体与能力层”。

接下来我们重点展开这两层，并给出可以直接改造的代码骨架。

---

三、智能体与能力层：把“能力”抽象成可组合的 Agent

3.1 定义一个统一的智能体接口

所有 Agent 无论是 HR、采购还是通用 RAG，都遵循同一个接口，便于中台调用和编排。

# agents/base_agent.py
from abc import ABC, abstractmethod
from typing import Dict, Any

class BaseAgent(ABC):
    """企业 AI 中台智能体统一接口"""

    @abstractmethod
    def meta(self) -> Dict[str, Any]:
        """返回智能体元信息，用于注册与发现"""
        # 例如：
        # {
        #   "name": "采购助手",
        #   "type": "procurement",
        #   "description": "处理采购相关需求与流程",
        #   "version": "1.0.0"
        # }
        pass

    @abstractmethod
    def can_handle(self, intent: str) -> bool:
        """根据意图判断是否适合处理该任务"""
        pass

    @abstractmethod
    def handle(self,
               task: Dict[str, Any],
               context: Dict[str, Any]) -> Dict[str, Any]:
        """
        处理任务的主入口
        :param task: 任务描述（由中台封装，含 intent / payload 等）
        :param context: 上下文（会话状态、多智能体共享信息）
        :return: 结果 + 对 context 的增量更新建议
        """
        pass

3.2 智能体注册中心：像注册微服务一样管理 Agent

# platform/agent_registry.py
from typing import Dict, List, Optional
from agents.base_agent import BaseAgent

class AgentRegistry:
    """AI 中台智能体注册中心"""

    def __init__(self):
        self._agents: Dict[str, BaseAgent] = {}

    def register(self, agent: BaseAgent):
        info = agent.meta()
        agent_type = info["type"]
        self._agents[agent_type] = agent
        print(f"[AgentRegistry] Registered agent: {agent_type} ({info['name']})")

    def get(self, agent_type: str) -> Optional[BaseAgent]:
        return self._agents.get(agent_type)

    def all(self) -> List[BaseAgent]:
        return list(self._agents.values())

    def match_by_intent(self, intent: str) -> List[BaseAgent]:
        """根据意图找到所有“宣称能处理”的智能体"""
        return [a for a in self._agents.values() if a.can_handle(intent)]

在应用启动时集中注册：

# main.py (部分)
from platform.agent_registry import AgentRegistry
from agents.procurement_agent import ProcurementAgent
from agents.hr_agent import HRAgent
from agents.rag_agent import RAGAgent

agent_registry = AgentRegistry()
agent_registry.register(ProcurementAgent())
agent_registry.register(HRAgent())
agent_registry.register(RAGAgent())

---

四、编排与协同层：中台的“大脑”

4.1 会话与任务状态管理

这里可以复用前几篇用到的 Redis / DB 方案，稍作抽象：

# platform/state_manager.py
import json
import redis
from typing import Dict, Any

class StateManager:
    """跨多智能体共享的会话 & 任务状态管理"""

    def __init__(self, redis_url="redis://localhost:6379/0"):
        self.r = redis.from_url(redis_url)
        self.ttl = 3600  # 1 小时

    def get_context(self, session_id: str) -> Dict[str, Any]:
        raw = self.r.get(f"ctx:{session_id}")
        return json.loads(raw) if raw else {}

    def save_context(self, session_id: str, ctx: Dict[str, Any]):
        self.r.setex(f"ctx:{session_id}", self.ttl, json.dumps(ctx))

state_manager = StateManager()

4.2 协调器（Orchestrator）：根据意图路由到合适的 Agent

# platform/orchestrator.py
import time
import uuid
from typing import Dict, Any, Optional, List
from platform.agent_registry import AgentRegistry
from platform.state_manager import state_manager
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

class Orchestrator:
    """企业 AI 中台的统一调度器"""

    def __init__(self, registry: AgentRegistry):
        self.registry = registry

    def infer_intent(self, message: str, session_ctx: Dict[str, Any]) -> str:
        """
        简化版：先用规则 + 关键词。  
        实际可替换为一个小模型/DeepSeek 调用做“意图识别”。
        """
        if any(k in message for k in ["采购", "买服务器", "报价", "下单"]):
            return "procurement.request"
        if any(k in message for k in ["请假", "报销", "年假"]):
            return "hr.policy"
        return "general.qa"

    def dispatch(self,
                 user_id: str,
                 session_id: str,
                 message: str) -> Dict[str, Any]:
        """入口：由 IM / Web / API 调用"""

        with tracer.start_as_current_span("orchestrator.dispatch") as span:
            span.set_attribute("user.id", user_id)
            span.set_attribute("session.id", session_id)

            # 1. 取上下文
            ctx = state_manager.get_context(session_id)

            # 2. 识别意图
            intent = self.infer_intent(message, ctx)
            span.set_attribute("intent", intent)

            # 3. 从注册中心找到候选智能体
            candidates = self.registry.match_by_intent(intent)
            if not candidates:
                # 回退到通用 RAG / 默认 Agent
                default_agent = self.registry.get("rag")
                candidates = [default_agent] if default_agent else []

            # 4. 按优先级/策略选一个主 Agent（可以后续扩展成多智能体协同）
            agent = candidates[0] if candidates else None
            if not agent:
                return {"answer": "当前暂不支持处理该类请求，请稍后再试。"}

            task = {
                "id": str(uuid.uuid4()),
                "intent": intent,
                "payload": {
                    "message": message,
                    "user_id": user_id
                },
                "timestamp": time.time()
            }

            # 5. 调用智能体
            with tracer.start_as_current_span(f"agent.{agent.meta()['type']}.handle"):
                result = agent.handle(task, ctx)

            # 6. 更新上下文
            new_ctx = result.get("context_updates", {})
            ctx.update(new_ctx)
            state_manager.save_context(session_id, ctx)

            return {
                "answer": result.get("answer", ""),
                "meta": {
                    "agent": agent.meta(),
                    "intent": intent,
                    "task_id": task["id"]
                }
            }

现在，所有入口（钉钉/企微/Web）都只需要调用：

orchestrator.dispatch(user_id, session_id, message)

而不用关心是哪个智能体在处理、是否会再转交其它智能体。

---

五、一个完整例子：采购 + HR + RAG 三智能体协同

下面通过一个具体场景串起来：
“员工发起采购需求 → 采购 Agent 拆解 + 查采购制度（RAG Agent） → HR Agent 校验预算归属”

5.1 采购 Agent：拆解需求 + 调用 RAG Agent 查制度

# agents/procurement_agent.py
from typing import Dict, Any
from agents.base_agent import BaseAgent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="deepseek-reasoner-671b", temperature=0.2)

PROMPT = ChatPromptTemplate.from_messages([
    ("system", "你是企业采购助手，请从用户描述中提取出采购金额、用途、紧急程度。输出 JSON。"),
    ("user", "{message}")
])

class ProcurementAgent(BaseAgent):

    def meta(self):
        return {
            "name": "采购助手",
            "type": "procurement",
            "description": "处理采购申请、解释采购流程"
        }

    def can_handle(self, intent: str) -> bool:
        return intent.startswith("procurement.")

    def handle(self, task: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
        message = task["payload"]["message"]

        # 1. 解析需求
        resp = llm.invoke(PROMPT.format_messages(message=message))
        import json
        info = json.loads(resp.content)

        amount = float(info.get("amount", 0))
        purpose = info.get("purpose", "")
        urgency = info.get("urgency", "normal")

        # 2. 调用“通用 RAG Agent”查采购制度（通过 context 协同）
        # 中台里我们约定：RAG Agent 会读 context["rag_query"] 字段
        context_updates = {
            "procurement": {
                "amount": amount,
                "purpose": purpose,
                "urgency": urgency
            },
            "rag_query": "公司采购审批规则，特别是关于金额区间和审批岗位的要求"
        }

        answer = (
            f"我已经理解你的采购需求：\n"
            f"- 金额：约 {amount} 元\n"
            f"- 用途：{purpose}\n"
            f"- 紧急程度：{urgency}\n\n"
            "接下来我会根据公司采购制度帮你确认需要走的审批流程。"
        )

        return {
            "answer": answer,
            "context_updates": context_updates,
            # 可选：提示中台下一步可以调哪个智能体
            "next_agents": ["rag", "hr"]
        }

5.2 通用 RAG Agent：基于 LlamaIndex 查制度文档

# agents/rag_agent.py
from typing import Dict, Any
from agents.base_agent import BaseAgent
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI

_docs = SimpleDirectoryReader("policies/").load_data()
_embed = OpenAIEmbedding(model_name="text-embedding-3-small")
_index = VectorStoreIndex.from_documents(_docs, embed_model=_embed)
_llm = OpenAI(model="deepseek-reasoner-671b")

class RAGAgent(BaseAgent):

    def meta(self):
        return {
            "name": "文档知识助手",
            "type": "rag",
            "description": "基于企业文档的 RAG 问答"
        }

    def can_handle(self, intent: str) -> bool:
        # 通用兜底
        return True

    def handle(self, task: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
        query = context.get("rag_query") or task["payload"]["message"]
        engine = _index.as_query_engine(similarity_top_k=3, llm=_llm)
        resp = engine.query(query)

        # 将结果写回 context，供其它智能体（如 HR、采购）使用
        context_updates = {
            "rag_last_answer": resp.response,
            "rag_last_sources": [n.node_id for n in resp.source_nodes]
        }

        return {
            "answer": resp.response,
            "context_updates": context_updates
        }

5.3 HR Agent：根据金额与岗位判定预算与审批责任

# agents/hr_agent.py
from typing import Dict, Any
from agents.base_agent import BaseAgent

class HRAgent(BaseAgent):

    def meta(self):
        return {
            "name": "人事与制度助手",
            "type": "hr",
            "description": "负责解释 HR / 组织 / 审批岗位相关规则"
        }

    def can_handle(self, intent: str) -> bool:
        return intent.startswith("hr.") or "hr.policy" in intent

    def handle(self, task: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
        procurement_info = context.get("procurement", {})
        amount = procurement_info.get("amount", 0)
        rag_policy = context.get("rag_last_answer", "")

        if not amount:
            answer = "目前还没有检测到具体的采购金额，无法为你匹配审批岗位。"
        else:
            # 简单规则示例
            if amount < 5000:
                approver = "部门负责人"
            elif amount < 20000:
                approver = "部门负责人 + 事业部 VP"
            else:
                approver = "部门负责人 + 事业部 VP + CEO"

            answer = (
                f"根据你当前的采购金额（约 {amount} 元），\n"
                f"结合公司采购制度，大致审批链路为：{approver}。\n\n"
                f"（制度摘录：{rag_policy[:150]}…）"
            )

        return {
            "answer": answer,
            "context_updates": {
                "procurement": {
                    **procurement_info,
                    "suggested_approver_chain": approver
                }
            }
        }

5.4 中台协同策略：串起多个 Agent

最简单的一步，可以在 Orchestrator 里做「多 Agent 顺序调用」：

# platform/orchestrator.py 里的 dispatch 增强版本（伪代码）
def dispatch(...):
    ...
    # 1. 主 Agent
    primary = candidates[0]
    result = primary.handle(task, ctx)
    ctx.update(result.get("context_updates", {}))

    # 2. 如果主 Agent 给出了 next_agents 列表，则继续调其它 Agent
    next_agents = result.get("next_agents", [])
    answers = [result["answer"]]

    for agent_type in next_agents:
        agent = self.registry.get(agent_type)
        if not agent:
            continue
        sub_result = agent.handle(task, ctx)
        ctx.update(sub_result.get("context_updates", {}))
        if sub_result.get("answer"):
            answers.append(sub_result["answer"])

    # 3. 汇总回答
    final_answer = "\n\n".join(answers)
    state_manager.save_context(session_id, ctx)
    return {"answer": final_answer, ...}

通过这种模式，你可以轻松实现：

• “采购 Agent + RAG Agent + HR Agent” 三者顺序协作
• 后续还可以拓展为并行调用、条件调用、循环调用等更复杂模式（可借助 LangGraph）

---

六、企业落地关键点：让中台“真能支撑多业务线”

最后讲几个在企业落地中常被忽视，但非常关键的点。

6.1 能力复用：把「公共能力」抽成中台服务

典型公共能力包括：

• 通用 RAG 查询（文档/制度/FAQ）
• 通用写作/总结（会议纪要、邮件草稿）
• 通知渠道（钉钉/企微/邮件/短信）
• 用户信息 & 权限查询（SSO / IAM）
• 审批流编排（通用审批引擎）

做法是：

• 不要在每个业务 Agent 里直接写这些逻辑
• 而是抽成**中台层的「通用 Agent/Tool」**​，所有业务 Agent 都调它们

6.2 权限与隔离：不是所有 Agent 都能访问所有数据

• 在中台层统一接企业 SSO / IAM，获得用户身份与角色
• 在调用某个 Agent 或工具前，做权限校验：
  • 某个部门的 Agent 只能访问本部门知识库和系统
  • 财务 Agent 访问的接口必须有更严格的审计与日志
• 对敏感操作（转账、删除、审批）做 “强制二次确认 + 人工兜底”

6.3 成本与配额：中台负责“算账”和“限流”

• 给不同业务线/部门设定 LLM 调用配额（按 token 或按调用次数）
• 在中台层统计：
  • 每个 Agent 的调用量、平均 token 数
  • 每个业务线的月度花费（可估算为 $/token * token 数）
• 对异常流量（突发高并发、超长输入）做限流与降级

6.4 监控与反馈：中台统一视角

把第（四）、（六）篇的能力整合进来：

• 每次中台调度都打上 Trace：看到从入口到每个 Agent/工具的完整链路
• 每个 Agent 的成功率、延迟、错误率，都能在同一张监控大盘上看
• 用户在任一场景给出的反馈（评分/吐槽），都能映射回某次中台调度 + 某个 Agent 的调用，从而精确定位问题

---

七、小结：让 AI 从“点上开花”到“面上铺开”

这一篇，我们完成了从「一个智能助手」到「多智能体协同的 AI 中台」的架构升级，关键动作包括：

1. 定义统一的 BaseAgent 接口 和 AgentRegistry 注册中心
2. 通过 Orchestrator + StateManager 把多智能体串联起来
3. 用一个完整的「采购 + HR + RAG」案例，演示了多智能体如何协同完成复杂任务
4. 从权限、成本、监控、反馈等多个维度，说明了 企业级中台 需要承担的职责

如果用一句话概括：

之前我们在做「一个很强的 AI 助手」，
现在我们开始做「一个能管理很多 AI 助手的操作系统」。