AI 协作平台的架构抉择:多 Agent 协同、上下文管理与工程落地

一、协作效率的悖论:为什么"AI + 团队"不等于"高效协作"

团队引入 AI 工具后,协作效率不升反降,这是一个被广泛忽视的现象。原因在于,当前大多数 AI 协作方案是"单点增强"而非"系统增强"——每个成员各自使用 AI 助手,但 AI 之间不通信,上下文不共享,决策不协同。

具体表现:产品经理用 AI 写需求文档,设计师用 AI 生成设计稿,开发用 AI 写代码——三个环节的 AI 各自为战,产品经理的 AI 不知道设计师的 AI 改了方案,开发的 AI 不知道需求已经更新。结果是人要花更多时间在 AI 之间做"翻译"和"对齐"。

真正的 AI 协作平台需要解决三个核心问题:第一,多 Agent 之间的上下文共享与同步;第二,人类在 AI 协作链路中的决策节点设计;第三,协作过程中的知识沉淀与复用。这三个问题不解决,AI 协作平台就只是"多个独立 AI 工具的拼盘"。

二、多 Agent 协作的架构模式

graph TB
    A[用户意图输入] --> B[Orchestrator 编排器]

    B --> C[需求分析 Agent]
    B --> D[设计评审 Agent]
    B --> E[代码审查 Agent]
    B --> F[测试生成 Agent]

    C --> G[共享上下文层]
    D --> G
    E --> G
    F --> G

    G --> H[知识库]
    G --> I[决策日志]

    B --> J{人工决策节点}
    J -->|批准| K[执行]
    J -->|修改| B
    J -->|拒绝| L[终止]

    style B fill:#e1f5fe
    style G fill:#e8f5e9
    style J fill:#fff3e0
    style H fill:#f3e5f5

2.1 编排器模式(Orchestrator)

编排器是多 Agent 协作的核心组件,负责将用户意图拆解为子任务、分配给合适的 Agent、协调 Agent 之间的依赖关系。编排器本身不做具体工作,只做调度和决策。

from dataclasses import dataclass, field
from typing import Optional, Callable
from enum import Enum
import uuid

class AgentRole(Enum):
    ANALYZER = "analyzer"       # 需求分析
    DESIGNER = "designer"       # 设计评审
    REVIEWER = "reviewer"       # 代码审查
    TESTER = "tester"           # 测试生成
    WRITER = "writer"           # 文档撰写

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    WAITING_HUMAN = "waiting_human"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class AgentTask:
    """Agent 任务单元"""
    task_id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
    role: AgentRole = None
    description: str = ""
    input_context: dict = field(default_factory=dict)
    output: Optional[dict] = None
    status: TaskStatus = TaskStatus.PENDING
    requires_human_approval: bool = False
    depends_on: list[str] = field(default_factory=list)  # 依赖的任务 ID

@dataclass
class SharedContext:
    """共享上下文层:所有 Agent 可读写的公共状态"""
    project_id: str
    requirements: dict = field(default_factory=dict)
    design_decisions: dict = field(default_factory=dict)
    code_changes: list[dict] = field(default_factory=list)
    review_comments: list[dict] = field(default_factory=list)
    decision_log: list[dict] = field(default_factory=list)

    def update(self, source_agent: str, key: str, value: dict) -> None:
        """更新上下文并记录变更日志"""
        setattr(self, key, value)
        self.decision_log.append({
            "agent": source_agent,
            "key": key,
            "timestamp": __import__("time").time(),
            "summary": f"{source_agent} 更新了 {key}",
        })

class Orchestrator:
    """多 Agent 编排器:负责任务拆解、调度和人工决策节点管理"""

    def __init__(self, context: SharedContext):
        self.context = context
        self.tasks: list[AgentTask] = []
        self.agent_registry: dict[AgentRole, Callable] = {}

    def register_agent(self, role: AgentRole, handler: Callable) -> None:
        """注册 Agent 处理函数"""
        self.agent_registry[role] = handler

    def plan(self, user_intent: str) -> list[AgentTask]:
        """将用户意图拆解为任务链"""
        # 根据意图类型生成任务序列
        task_chain = [
            AgentTask(
                role=AgentRole.ANALYZER,
                description=f"分析用户意图: {user_intent}",
                requires_human_approval=True,  # 需求分析需人工确认
            ),
            AgentTask(
                role=AgentRole.DESIGNER,
                description="基于需求生成技术方案",
                depends_on=[self.tasks[0].task_id if self.tasks else ""],
                requires_human_approval=True,  # 方案需人工审批
            ),
            AgentTask(
                role=AgentRole.REVIEWER,
                description="审查代码变更",
                depends_on=[self.tasks[1].task_id if len(self.tasks) > 1 else ""],
            ),
            AgentTask(
                role=AgentRole.TESTER,
                description="生成测试用例",
                depends_on=[self.tasks[2].task_id if len(self.tasks) > 2 else ""],
            ),
        ]
        self.tasks = task_chain
        return task_chain

    def execute_next(self) -> Optional[dict]:
        """执行下一个可运行的任务"""
        for task in self.tasks:
            if task.status != TaskStatus.PENDING:
                continue

            # 检查依赖是否完成
            deps_met = all(
                any(t.task_id == dep_id and t.status == TaskStatus.COMPLETED
                    for t in self.tasks)
                for dep_id in task.depends_on
                if dep_id  # 跳过空依赖
            )
            if not deps_met:
                continue

            # 执行 Agent
            handler = self.agent_registry.get(task.role)
            if not handler:
                task.status = TaskStatus.FAILED
                continue

            task.status = TaskStatus.RUNNING
            result = handler(task.input_context, self.context)

            if task.requires_human_approval:
                task.status = TaskStatus.WAITING_HUMAN
                return {
                    "action": "human_approval_needed",
                    "task_id": task.task_id,
                    "result": result,
                }

            task.output = result
            task.status = TaskStatus.COMPLETED

        return {"action": "all_completed"}

2.2 共享上下文层的设计

共享上下文是多 Agent 协作的关键基础设施。它解决了"Agent 之间如何知道彼此做了什么"的问题。设计原则是:所有 Agent 只通过共享上下文通信,不直接调用其他 Agent。

import json
from datetime import datetime

class ContextStore:
    """持久化上下文存储:支持版本控制和变更追踪"""

    def __init__(self, project_id: str):
        self.project_id = project_id
        self._versions: list[dict] = []
        self._current: dict = {
            "requirements": {},
            "design": {},
            "code": {},
            "reviews": [],
            "decisions": [],
        }

    def write(self, agent_id: str, section: str, data: dict) -> None:
        """Agent 写入上下文"""
        # 记录变更前的快照
        snapshot = json.dumps(self._current, ensure_ascii=False)

        self._current[section] = data
        self._current["decisions"].append({
            "agent": agent_id,
            "section": section,
            "timestamp": datetime.utcnow().isoformat(),
            "action": "update",
        })

        # 保存版本快照,支持回溯
        self._versions.append({
            "version": len(self._versions) + 1,
            "snapshot": snapshot,
            "changed_by": agent_id,
            "timestamp": datetime.utcnow().isoformat(),
        })

    def read(self, section: str) -> dict:
        """Agent 读取上下文"""
        return self._current.get(section, {})

    def get_changes_since(self, version: int) -> list[dict]:
        """获取指定版本以来的所有变更"""
        return [v for v in self._versions if v["version"] > version]

    def rollback(self, version: int) -> None:
        """回滚到指定版本"""
        if 0 < version <= len(self._versions):
            self._current = json.loads(
                self._versions[version - 1]["snapshot"]
            )

2.3 人工决策节点

AI 协作平台不能是完全自动化的。在关键决策点(需求确认、方案审批、上线决策)必须设置人工审批节点。这不仅是安全要求,更是效率要求——让 AI 在错误方向上高速运转,比没有 AI 更危险。

@dataclass
class HumanApprovalGate:
    """人工审批门:在关键决策点暂停自动化流程"""

    task_id: str
    agent_role: AgentRole
    decision_summary: str          # AI 生成的决策摘要
    risk_level: str                # LOW / MEDIUM / HIGH
    auto_approve_threshold: str = "LOW"  # 低于此风险等级可自动通过

    def should_auto_approve(self) -> bool:
        """判断是否可以自动通过"""
        risk_order = {"LOW": 0, "MEDIUM": 1, "HIGH": 2}
        return risk_order.get(self.risk_level, 2) <= \
               risk_order.get(self.auto_approve_threshold, 0)

    def format_approval_request(self) -> str:
        """格式化审批请求,供人类决策者审阅"""
        return (
            f"[审批请求] 任务 {self.task_id}\n"
            f"来源: {self.agent_role.value} Agent\n"
            f"风险等级: {self.risk_level}\n"
            f"决策摘要: {self.decision_summary}\n"
            f"操作: 批准(y) / 修改后批准(m) / 拒绝(n)"
        )

三、AI 协作平台的技术实现

3.1 基于 Event-Driven 的异步协作架构

import asyncio
from collections import defaultdict

class EventBus:
    """事件总线:Agent 之间的异步通信通道"""

    def __init__(self):
        self._subscribers: dict[str, list[asyncio.Queue]] = defaultdict(list)

    async def publish(self, event_type: str, payload: dict) -> None:
        """发布事件"""
        for queue in self._subscribers[event_type]:
            await queue.put(payload)

    def subscribe(self, event_type: str) -> asyncio.Queue:
        """订阅事件,返回异步队列"""
        queue = asyncio.Queue()
        self._subscribers[event_type].append(queue)
        return queue

# 使用示例:Agent 之间通过事件总线协作
bus = EventBus()

async def code_review_agent():
    """代码审查 Agent:监听代码提交事件"""
    queue = bus.subscribe("code_committed")
    while True:
        commit = await queue.get()
        review_result = await review_code(commit)

        if review_result["issues"]:
            # 发现问题,发布审查事件
            await bus.publish("review_completed", {
                "commit_id": commit["id"],
                "status": "issues_found",
                "issues": review_result["issues"],
            })
        else:
            await bus.publish("review_completed", {
                "commit_id": commit["id"],
                "status": "approved",
            })

async def test_agent():
    """测试 Agent:监听审查通过事件"""
    queue = bus.subscribe("review_completed")
    while True:
        review = await queue.get()
        if review["status"] == "approved":
            tests = await generate_tests(review["commit_id"])
            await bus.publish("tests_generated", {
                "commit_id": review["commit_id"],
                "test_count": len(tests),
            })

3.2 知识沉淀与复用

class KnowledgeBase:
    """知识库:沉淀协作过程中的决策和经验"""

    def __init__(self):
        self._entries: list[dict] = []

    def record_decision(self, context: str, decision: str,
                        rationale: str, outcome: Optional[str] = None) -> None:
        """记录决策及其理由"""
        self._entries.append({
            "type": "decision",
            "context": context,
            "decision": decision,
            "rationale": rationale,
            "outcome": outcome,
            "timestamp": datetime.utcnow().isoformat(),
        })

    def record_pattern(self, problem: str, solution: str,
                       applicability: str) -> None:
        """记录可复用的解决模式"""
        self._entries.append({
            "type": "pattern",
            "problem": problem,
            "solution": solution,
            "applicability": applicability,
            "timestamp": datetime.utcnow().isoformat(),
        })

    def search_relevant(self, query: str, top_k: int = 5) -> list[dict]:
        """检索相关知识(简化版,生产环境应使用向量检索)"""
        # 按关键词匹配排序
        scored = []
        query_terms = set(query.lower().split())
        for entry in self._entries:
            text = f"{entry.get('context', '')} {entry.get('problem', '')} " \
                   f"{entry.get('solution', '')} {entry.get('rationale', '')}"
            text_terms = set(text.lower().split())
            overlap = len(query_terms & text_terms)
            if overlap > 0:
                scored.append((overlap, entry))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [entry for _, entry in scored[:top_k]]

四、AI 协作平台的边界与风险

上下文膨胀问题。随着协作流程推进,共享上下文的数据量不断增长。如果每个 Agent 都读取完整上下文,Token 消耗会指数级增长。解决方案是按需裁剪上下文:每个 Agent 只读取与当前任务相关的上下文片段,而非全量上下文。但裁剪策略本身是一个难题——裁剪过度会导致 Agent 丢失关键信息,裁剪不足则浪费 Token。

Agent 间的级联失败。一个 Agent 的错误输出会成为下一个 Agent 的输入,导致错误在链路中放大。例如,需求分析 Agent 理解错了用户意图,设计 Agent 基于错误需求生成了方案,代码审查 Agent 基于错误方案审查了代码——三层放大后,最终产出与用户意图完全偏离。缓解方案是在每个 Agent 的输出中增加置信度评分,低置信度的输出自动触发人工审批。

知识库的噪声积累。知识库中的记录质量参差不齐,过时的决策模式会误导后续 Agent。必须建立知识库的淘汰机制:超过 6 个月未被引用的条目降权,与当前项目上下文不匹配的条目标记为"可能过时"。

人工审批疲劳。如果每个决策点都需要人工审批,审批者会因决策过多而降低审批质量。建议将自动审批阈值设为 LOW,只对 HIGH 风险决策强制人工审批,MEDIUM 风险决策采用"默认通过 + 事后审计"模式。

五、总结

AI 协作平台的核心挑战不是单个 Agent 的能力,而是多 Agent 之间的协同效率。编排器模式、共享上下文层和人工决策节点是解决协同问题的三大支柱。

落地路线建议:第一步,明确协作流程中的 Agent 角色和任务依赖关系,用编排器统一调度;第二步,建立共享上下文层,所有 Agent 只通过上下文通信,避免直接耦合;第三步,在需求确认、方案审批和上线决策三个关键节点设置人工审批门;第四步,实现事件驱动的异步架构,Agent 之间通过事件总线解耦;第五步,建立知识库的沉淀和淘汰机制,确保知识质量不随时间退化。

AI 协作的目标不是替代人,而是让人的决策更高效、更准确。任何试图完全自动化的协作系统,最终都会因为缺乏人的判断力而失败。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐