aiagent/docs/architecture/解决缺失能力计划.md

解决缺失能力计划

总体策略

按梯队分阶段实施。每个阶段聚焦 2-3 个能力，完成即可独立提升系统质量。 每阶段包含：涉及文件、核心改动、验证方式。

---

第一阶段：闭环质量（1-2 周）

优先解决"输出质量验证"，这是连接自主进化系统的最后一环。

1.1 输出质量验证 → 新增 evaluator 节点 + self_review 工具

目标： Agent 执行完后自动检查输出质量，不达标则自我修正。

方案：

A. 新增 self_review 工具（第 35 个内置工具）

self_review(content="...", criteria="回答必须包含SQL优化建议和执行计划")
→ {
    "score": 0.75,
    "passed": true,
    "issues": ["缺少具体执行计划"],
    "suggestions": ["补充EXPLAIN输出分析"]
  }

实现： 用 LLM 作为评判器（deepseek-v4-flash，轻量调用），根据 criteria 打分 0-1。

B. 新增 evaluator 工作流节点类型

在工作流引擎中添加 evaluator 节点：

• 接收上游节点输出 + 评判标准（criteria）
• 调用 LLM 评判输出质量
• 输出 {score, passed, issues, suggestions}
• 配合 condition 节点：passed=false → 走修正分支 → 重试

C. AgentRuntime 内建 self-review

在 AgentRuntime ReAct 循环末尾添加可选的 self-review 步骤：

• AgentConfig 添加 self_review: bool = False
• 启用后，Agent 在返回最终答案前自动调用 self_review
• 评分 < 0.6 则追加一轮修正迭代

涉及文件：

文件	改动
backend/app/services/builtin_tools.py	新增 self_review_tool + SELF_REVIEW_SCHEMA
backend/app/core/tools_bootstrap.py	34→35，import + register
backend/app/services/workflow_engine.py	新增 evaluator 节点类型
backend/app/agent_runtime/core.py	末尾添加 self-review 步骤
backend/app/agent_runtime/schemas.py	AgentConfig 添加 self_review 字段
frontend/src/utils/agentSkills.ts	新增条目

验证：

1. self_review("SELECT * FROM t", "需要包含索引建议") → score < 0.5
2. 工作流：agent → evaluator(criteria="...") → condition(passed?) → 正常结束 / 修正重试
3. 全能助手对话测试：问一个复杂问题，开启 self_review，确认输出质量提升

---

1.2 节点级自动重试

目标： error_handler 从空壳变为真正的重试机制。

方案：

修改 error_handler 节点执行逻辑：

当上游节点 status == "failed" 时：
  1. 检查 retry_count（剩余重试次数）
  2. 等待 retry_delay 秒
  3. 重新执行上游节点（复用 get_node_input + execute_node）
  4. 成功 → 继续流程
  5. 失败且 retry_count > 0 → 循环
  6. 失败且 retry_count == 0 → 触发 on_error 动作

on_error 动作：

• stop：抛出错误停止工作流
• notify：发送告警继续
• skip：跳过该节点继续

涉及文件：

文件	改动
backend/app/services/workflow_engine.py	重写 error_handler 节点逻辑
backend/app/core/exceptions.py	添加 NodeRetryExhausted 异常

验证：

1. 模拟 LLM 超时 → error_handler 重试 3 次 → 第 3 次成功 → 继续
2. 模拟连续失败 → error_handler 耗尽重试 → 触发 on_error=notify → 告警记录

---

第二阶段：编排整合（2-3 周）

2.1 Orchestrator 进入工作流

目标： 4 种编排模式变为工作流节点类型。

方案：

新增 orchestrator 工作流节点类型，数据配置：

{
  "mode": "route | sequential | debate | pipeline",
  "agents": ["agent_id_1", "agent_id_2", ...],
  "routing_prompt": "用于 route 模式的路由指令",
  "aggregation_prompt": "用于 debate 模式的汇总指令"
}

workflow_integration.py 添加 run_orchestrator_node()：

• 解析 node_data 中的 agent 列表
• 从 DB 加载 Agent 配置
• 创建 AgentOrchestrator 实例
• 执行对应模式，返回结构化结果

涉及文件：

文件	改动
backend/app/agent_runtime/workflow_integration.py	新增 run_orchestrator_node()
backend/app/services/workflow_engine.py	新增 orchestrator 节点类型
frontend/src/utils/agentSkills.ts	无需改动（后端节点类型）

验证：

1. 创建工作流：start → orchestrator(debate, [AgentA, AgentB, AgentC]) → end
2. 执行：确认 3 个 Agent 均被调用，结果被汇总
3. 创建工作流：start → orchestrator(route, [代码助手, 数据分析师]) → end，确认按路由分发

---

2.2 工具级人工审批

目标： 危险工具执行前暂停等待用户确认。

方案：

A. AgentToolConfig 扩展

class AgentToolConfig(BaseModel):
    include_tools: List[str] = []
    exclude_tools: List[str] = []
    require_approval: List[str] = []  # 新增：需要审批的工具名列表

B. AgentRuntime 审批拦截

在 ReAct 循环的工具执行环节（core.py 约 287 行）：

if tool_name in config.tools.require_approval:
    yield ApprovalRequest(tool_name, tool_args)  # 暂停，等待外部确认
    decision = await wait_for_approval()  # 阻塞等待
    if decision != "approved":
        continue  # 跳过该工具调用

C. WebSocket 审批通道

• AgentRuntime 暂停时通过 WebSocket 推送审批请求
• 用户通过 WebSocket 回复 approve/deny
• 超时默认策略可配置（approve/deny/skip）

涉及文件：

文件	改动
backend/app/agent_runtime/schemas.py	AgentToolConfig 添加 require_approval
backend/app/agent_runtime/core.py	工具执行前审批拦截
backend/app/websocket/manager.py	添加审批消息类型

验证：

1. 配置 Agent: require_approval: ["deploy_push"]
2. Agent 调用 deploy_push → WebSocket 收到审批请求
3. 用户 approve → Agent 继续执行
4. 用户 deny → Agent 跳过该工具

---

第三阶段：性能与效率（2-3 周）

3.1 并行执行

目标： 无依赖节点并行执行，Orchestrator debate 并行。

方案：

A. 工作流 DAG 并行

修改 execute() 主循环：

当前：每轮取 1 个节点执行
改为：每轮取所有"前置节点全部完成"的节点，asyncio.gather 并行执行

注意事项：

• 并行节点不能有共享状态依赖
• 需要分别捕获每个并行节点的异常
• WebSocket 需支持多节点同时推送进度

B. Orchestrator debate 并行

# 改前
for agent in agents:
    result = await agent.run(query)

# 改后
results = await asyncio.gather(*[agent.run(query) for agent in agents])

涉及文件：

文件	改动
backend/app/services/workflow_engine.py	execute() 并行执行
backend/app/agent_runtime/orchestrator.py	debate/sequential 并行
backend/app/websocket/manager.py	支持多节点并发推送

验证：

1. 3 个独立分支的 DAG → 总耗时 ≈ 最长分支耗时（而非三者之和）
2. debate 模式 3 Agent → 总耗时 ≈ 最慢 Agent 耗时

---

3.2 粒度进度上报

目标： 实时推送"第 3/7 步完成"到前端。

方案：

A. WorkflowEngine 进度回调

在 execute() 循环中添加钩子：

async def on_progress(self, current_step, total_steps, node_name, status):
    # 推送到 WebSocket
    await ws_manager.broadcast(execution_id, {
        "type": "progress",
        "current": current_step,
        "total": total_steps,
        "node": node_name,
        "status": status,
        "percent": int(current_step / total_steps * 100),
    })

B. 统计 DAG 总步数

执行前遍历拓扑排序结果，计算可达节点总数作为 total_steps。

C. WebSocket 改为推送模式

当前是轮询 DB，改为 WorkflowEngine 主动调用 WebSocket manager 推送。

涉及文件：

文件	改动
backend/app/services/workflow_engine.py	添加 on_progress 回调 + 步数统计
backend/app/websocket/manager.py	添加 execution 进度推送方法
backend/app/api/websocket.py	适配新的推送模式

验证：

1. 执行 7 节点的 DAG → WebSocket 收到 7 条进度消息
2. 前端进度条从 0% 逐步递增到 100%

---

3.3 结果缓存

目标： 相同输入不重复计算。

方案：

添加缓存层，使用 Redis：

# 工具结果缓存
cache_key = f"tool:{tool_name}:{hash(json.dumps(args))}"
cached = await redis.get(cache_key)
if cached:
    return cached.decode()

# LLM 响应缓存（相同 prompt + 相同 messages）
cache_key = f"llm:{hash(messages_json)}"

配置：

• TTL 默认 1 小时
• 工具维度可选开启/关闭
• 确定性工具（file_read、math_calculate）默认开启
• 非确定性工具（web_search、http_request）默认关闭

涉及文件：

文件	改动
backend/app/agent_runtime/tool_manager.py	工具执行前后加缓存层
backend/app/agent_runtime/core.py	LLM 调用前后加缓存层
backend/app/core/config.py	添加缓存配置项

验证：

1. 连续两次相同 file_read → 第二次命中缓存（耗时 < 1ms）
2. 连续两次相同 math_calculate → 第二次命中缓存

---

第四阶段：容错与共享（2-3 周）

4.1 降级/回退链

目标： 模型/Agent 失败自动切换备用方案。

方案：

AgentLLMConfig 扩展：

class AgentLLMConfig(BaseModel):
    provider: str = "openai"
    model: str = "gpt-4o-mini"
    fallback_llm: Optional['AgentLLMConfig'] = None  # 降级模型

Agent 节点 data 扩展：

{
  "fallback_agent": "备选Agent的ID"  // 可选
}

执行逻辑：

1. 主模型调用失败 → 切换 fallback_llm 重试
2. Agent 执行失败 → 查找 fallback_agent 重试
3. 所有备用方案都失败 → 抛出错误

涉及文件：

文件	改动
backend/app/agent_runtime/schemas.py	AgentLLMConfig 添加 fallback_llm
backend/app/agent_runtime/core.py	LLM 调用失败后切换 fallback
backend/app/services/workflow_engine.py	Agent 节点失败后切换 fallback_agent

---

4.2 Agent 间知识共享

目标： 打破记忆隔离，Agent 间共享知识。

方案：

添加全局知识索引层：

每个 Agent 执行完成后 → 提取关键知识 → 写入 global_knowledge 表
Agent 初始化时 → 从 global_knowledge 加载相关条目

实现：

• 新增 GlobalKnowledge 模型（content, embedding, source_agent_id, tags, created_at）
• AgentMemory.initialize() 添加全局知识检索步骤
• 自主进化创建的 Agent 自动继承创建者的知识

涉及文件：

文件	改动
backend/app/models/agent.py	新增 GlobalKnowledge 模型
backend/app/agent_runtime/memory.py	initialize() 添加全局知识加载
backend/app/agent_runtime/core.py	执行完毕提取知识写入全局

---

4.3 Agent 异步执行实现

目标： 填空 execute_agent_task，支持真正异步 Agent 执行。

方案：

execute_agent_task 当前是空壳，需要实现：

@celery_app.task
def execute_agent_task(agent_id, query, user_id):
    db = SessionLocal()
    agent = db.query(Agent).filter(Agent.id == agent_id).first()
    config = build_agent_config_from_db(agent)
    runtime = AgentRuntime(config)
    result = asyncio.run(runtime.run(query))
    # 更新 execution 记录
    return {"status": "completed", "output": result.content}

涉及文件：

文件	改动
backend/app/tasks/agent_tasks.py	实现 execute_agent_task
backend/app/tasks/scheduler_tasks.py	调度触发真正的异步执行

---

实施路线图总览

第一阶段（1-2周）
  ├── 1.1 输出质量验证：self_review工具 + evaluator节点 + AgentRuntime自检
  └── 1.2 节点级自动重试：error_handler从空壳到真正重试

第二阶段（2-3周）
  ├── 2.1 Orchestrator进入工作流：新增orchestrator节点类型
  └── 2.2 工具级人工审批：AgentToolConfig审批标记 + WebSocket审批通道

第三阶段（2-3周）
  ├── 3.1 并行执行：DAG并行 + Orchestrator debate并行
  ├── 3.2 粒度进度上报：on_progress回调 + WebSocket推送
  └── 3.3 结果缓存：Redis缓存层

第四阶段（2-3周）
  ├── 4.1 降级/回退链：fallback_llm + fallback_agent
  ├── 4.2 Agent知识共享：GlobalKnowledge + 跨Agent检索
  └── 4.3 Agent异步执行：填空execute_agent_task

优先级逻辑

• 第一阶段优先：质量验证是自主进化的最后一环，没有它进化无方向；重试是稳定性基础
• 第二阶段其次：编排整合让复杂任务可视化、安全化
• 第三阶段提速：并行 + 缓存大幅提升性能（生产部署前必备）
• 第四阶段收尾：容错和共享让系统长期运行可持续