智能体系统中的Controller/Worker架构设计与实践

markdown复制## 1. 项目概述：Controller/Worker模式在智能体系统中的工程化实践

这个项目展示了一个典型的企业级智能体系统实现方案。不同于常见的单体式AI应用，我们采用Controller/Worker架构将复杂任务分解为可独立管理的子模块。这种设计源于我在多个AI项目落地过程中积累的深刻教训——当系统需要同时处理SQL查询、文档检索和结果合成时，传统的端到端模型调用方式很快就会变得难以维护和审计。

### 1.1 核心需求解析

现代企业知识管理通常面临三类典型问题：
1. **结构化数据查询**：如"新加坡办公室有多少员工"这类可通过数据库直接回答的事实性问题
2. **非结构化文档检索**：如"安全策略中的最小权限原则是什么"这类需要从文档中提取的规则性问题
3. **混合型问题**：如"列出所有P0工单并说明对应的升级流程"这类需要同时结合数据和文档的复合问题

传统做法是直接将问题抛给大语言模型处理，但这会带来三个严重问题：
- 无法确保模型始终选择正确的信息源
- 缺乏对敏感操作（如数据库查询）的安全控制
- 出现问题难以追踪故障点

### 1.2 架构设计理念

我们的解决方案基于以下核心原则：
- **职责分离**：路由决策、SQL执行、文档检索、结果合成分别由不同Worker处理
- **显式控制流**：Controller明确控制执行流程，而非依赖模型的隐式推理
- **结构化中间产物**：各阶段产出标准化数据结构，便于追踪和审计
- **安全边界**：关键操作（如SQL执行）有严格的校验机制

## 2. 核心组件实现细节

### 2.1 系统分层架构

#### 控制层（Controller）
作为系统大脑，Controller不处理具体业务逻辑，而是负责：
- 初始化执行上下文（Trace ID、问题解析）
- 按顺序调用Worker（路由→取证→合成）
- 收集和持久化执行轨迹
- 处理异常情况和空结果回退

典型代码结构：
```python
def handle(question_text):
    # 初始化上下文
    ctx = TaskContext(question=question_text)
    
    # 阶段1：路由决策
    route_result = router_worker.run(ctx)
    
    # 阶段2：证据收集
    evidence = EvidenceSet()
    if route_result.needs_sql:
        evidence.extend(sql_worker.run(ctx))
    if route_result.needs_rag:
        evidence.extend(rag_worker.run(ctx))
    
    # 阶段3：结果合成
    ctx.state['evidence'] = evidence
    answer = synthesis_worker.run(ctx)
    
    # 产物输出
    return AnswerPackage(answer, ctx.state['citations'])

执行层（Workers）

每个Worker都是独立的功能单元：

1. QueryRouterWorker：

   • 使用LLM判断问题类型（SQL/RAG/MIX）
   • 包含启发式回退机制
   • 输出标准化路由决策对象

2. SQLWorker：

   • 生成符合业务语义的SQL
   • 执行严格的SQL安全校验
   • 将查询结果转换为结构化证据

3. RAGWorker：

   • 文档分块和向量化处理
   • 基于相似度的TopK检索
   • 结果标准化为证据项

4. SynthesisWorker：

   • 仅基于提供的证据生成答案
   • 自动提取引用标记
   • 处理空证据情况

数据层（Models）

定义系统核心数据结构：

• TaskContext：执行上下文，包含问题、状态字典等
• WorkerResult：标准化Worker输出
• EvidenceItem：统一证据表示（SQL结果/文档片段）
• AnswerPackage：最终输出容器

2.2 关键实现技术

SQL安全防护机制

为防止SQL注入和过度查询，我们实现了一套校验规则：

python复制def validate_sql(sql, allowed_tables):
    # 只允许SELECT语句
    if not sql.strip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT statements allowed")
    
    # 禁止多语句执行
    if ";" in sql:
        raise ValueError("Multiple statements prohibited")
    
    # 检查表名白名单
    used_tables = extract_tables(sql)
    for table in used_tables:
        if table.lower() not in [t.lower() for t in allowed_tables]:
            raise ValueError(f"Table {table} not allowed")
    
    # 强制添加LIMIT子句
    if "LIMIT" not in sql.upper():
        sql = f"{sql.rstrip(';')} LIMIT 50"
    
    return sql

文档检索优化

文档处理流程包含以下优化点：

1. 预处理阶段：

   • 按语义边界分块（如Markdown标题）
   • 添加元数据（文档名、章节等）

2. 检索阶段：

   • 使用稠密向量检索（Dense Retrieval）
   • 支持多语言查询
   • 结果按相关性排序

3. 后处理阶段：

   • 证据去重
   • 相关性阈值过滤
   • 上下文片段合并

执行追踪（Tracing）

系统记录完整的执行轨迹：

json复制{
  "events": [
    {
      "type": "route",
      "data": {
        "decision": "MIX",
        "confidence": 0.85,
        "rationale": "问题需要同时查询工单数据和流程文档"
      }
    },
    {
      "type": "sql",
      "data": {
        "query": "SELECT * FROM tickets WHERE priority='P0' LIMIT 50",
        "row_count": 3
      }
    }
  ]
}

3. 典型问题处理流程

3.1 纯SQL问题："新加坡有多少员工？"

处理流程：

1. Router判断为SQL类型
2. SQLWorker生成查询：

   sql复制SELECT COUNT(*) FROM employees 
   WHERE office='新加坡' OR office='Singapore'
   

3. 直接返回查询结果

3.2 纯RAG问题："最小权限原则是什么？"

处理流程：

1. Router判断为RAG类型
2. RAGWorker检索相关文档片段
3. SynthesisWorker基于检索结果生成答案

3.3 混合问题："列出P0工单及升级流程"

处理流程：

1. Router判断为MIX类型
2. 并行执行：
   • SQLWorker查询P0工单
   • RAGWorker检索升级流程文档
3. 合并证据后生成综合答案

4. 工程实践中的经验教训

4.1 必须避免的陷阱

1. 过度依赖模型自觉性：

   • 错误做法：期望模型自行决定何时查数据库、何时检索文档
   • 正确做法：通过Router显式控制流程

2. 忽视中间产物标准化：

   • 错误做法：各Worker返回自由格式结果
   • 正确做法：定义统一的EvidenceItem结构

3. 缺乏执行追踪：

   • 错误做法：只记录最终答案
   • 正确做法：记录每个决策点和操作结果

4.2 性能优化建议

1. 并行化取证阶段：

python复制# 使用ThreadPoolExecutor并行执行
with ThreadPoolExecutor() as executor:
    sql_future = executor.submit(sql_worker.run, ctx) if needs_sql else None
    rag_future = executor.submit(rag_worker.run, ctx) if needs_rag else None
    
    evidence = EvidenceSet()
    if sql_future: evidence.extend(sql_future.result())
    if rag_future: evidence.extend(rag_future.result())

2. 缓存策略：

   • 缓存频繁查询的SQL结果
   • 缓存文档嵌入向量
   • 实现查询结果LRU缓存

3. 索引优化：

   • 为常用查询字段建立数据库索引
   • 文档分块时保留结构信息
   • 使用专业向量数据库处理大规模文档

5. 扩展与演进路线

5.1 短期改进方向

1. 增强Router能力：

   • 支持多步计划生成
   • 集成业务规则引擎
   • 添加验证反馈机制

2. 证据后处理：

   • 实现证据去重
   • 添加冲突消解逻辑
   • 引入可信度评估

5.2 中长期规划

1. 系统集成：

   • 对接企业CMDB系统
   • 集成工单管理系统
   • 连接监控告警平台

2. 运维增强：

   • 基于trace的自动化测试
   • 性能指标监控
   • 自动回归测试框架

3. 知识管理：

   • 实现文档增量更新
   • 支持多版本知识库
   • 添加知识新鲜度检测

6. 实施建议与资源规划

6.1 团队技能要求

• 熟悉Python异步编程
• 掌握基础SQL优化技巧
• 了解向量检索原理
• 具备基础的DevOps能力

6.2 硬件资源配置

6.3 实施里程碑

1. 第一阶段（1个月）：核心流程实现
2. 第二阶段（2周）：安全加固和性能优化
3. 第三阶段（2周）：监控和运维工具开发
4. 第四阶段（持续）：知识库建设和迭代

在实际部署过程中，我们建议采用渐进式演进策略，先从关键业务场景入手，验证架构可行性后再逐步扩展功能范围。同时要特别注意建立完善的质量评估体系，包括自动化测试套件和人工评估流程，确保系统迭代过程中核心能力不会退化。