MBE Agent 框架文档

多智能体协作编排 · MCP Tool 框架 · 智能路由 — 面向开发者的 Agent 集成指南

目录

• 1. 架构概览
• 2. 多智能体编排器
• 3. MCP 协议与工具
• 4. 智能路由
• 5. SCST 自进化训练
• 6. 因果推理引擎
• 7. 离线 CoT 框架
• 8. API 参考
• 9. 最佳实践

1. 架构概览

MBE 的 Agent 框架建立在 Level 6（自主决策智能体）能力之上，由七大核心模块协同工作：

┌────────────────────────────────────────────────────────────┐
│                    用户请求                                  │
├────────────────────────────────────────────────────────────┤
│                                                            │
│   ┌──────────┐    ┌──────────────┐    ┌──────────────┐    │
│   │ 智能路由  │───→│ 多智能体编排  │───→│ 回答融合      │    │
│   │SmartRouter│    │ Orchestrator │    │ Fusion       │    │
│   └──────────┘    └──────────────┘    └──────────────┘    │
│        ↑                 ↑                    ↓            │
│   ┌──────────┐    ┌──────────────┐    ┌──────────────┐    │
│   │ 因果推理  │    │ 知识图谱      │    │ Self-Critique │    │
│   │ Causal   │    │ Knowledge    │    │ 一致性验证    │    │
│   └──────────┘    └──────────────┘    └──────────────┘    │
│                                                            │
│   ┌──────────┐    ┌──────────────┐    ┌──────────────┐    │
│   │ MCP 工具  │    │ SCST 训练    │    │ 离线 CoT     │    │
│   │ Tools    │    │ Self-Train   │    │ Reasoning    │    │
│   └──────────┘    └──────────────┘    └──────────────┘    │
└────────────────────────────────────────────────────────────┘

能力层级

2. 多智能体编排器

2.1 概述

AgentOrchestrator 实现了基于知识图谱驱动的多专家协作框架。

核心流程：

analyze_domains() → should_multi_agent() → select_agents() → parallel_query() → fuse_responses()
     分析领域            判断是否多Agent         选择专家          并行调用            融合回答

代码位置：src/core/agent_orchestrator.py

2.2 核心类

AgentOrchestrator

from src.core.agent_orchestrator import get_agent_orchestrator

orchestrator = get_agent_orchestrator(
    max_agents=3,                    # 最多同时调用的专家数
    min_confidence=0.3,              # 专家最低置信度
    consistency_threshold=0.6,       # 一致性阈值
)

完整编排调用

result = await orchestrator.orchestrate(
    question="我想投资新能源，同时考虑法律风险和税务优化",
    system_prompt="请从专业角度分析",
    force_multi=False,               # 是否强制多Agent
)

# result: OrchestratedResult
print(result.is_multi_agent)         # True
print(result.agents_used)            # ["finance_expert", "legal_expert"]
print(result.fused_answer)           # 融合后的综合回答
print(result.decision_reason)        # "问题涉及 2 个领域: finance, legal"

2.3 数据模型

@dataclass
class AgentResponse:
    agent_id: str           # 专家 ID
    agent_name: str         # 专家名称
    domain: str             # 专业领域
    answer: str             # 回答内容
    confidence: float       # 置信度 [0, 1]
    sources: List[str]      # 引用来源
    latency_ms: float       # 响应延迟

@dataclass
class OrchestratedResult:
    question: str                              # 原始问题
    is_multi_agent: bool                       # 是否使用多Agent
    agents_used: List[str]                     # 参与的专家
    individual_responses: List[AgentResponse]   # 各专家回答
    fused_answer: str                          # 融合后的最终回答
    consistency_score: float                   # 跨Agent一致性
    total_latency_ms: float                    # 总延迟
    decision_reason: str                       # 使用多Agent的原因

2.4 多Agent 触发条件

编排器通过以下规则判断是否需要多 Agent：

3. MCP 协议与工具

3.1 概述

MBE 实现了完整的 MCP（Model Context Protocol）服务器，支持通过 JSON-RPC over SSE 协议与外部系统集成。

代码位置：src/mcp/

src/mcp/
├── server.py       # MCP 服务器（JSON-RPC 处理）
├── tools.py        # 工具定义（7 个内置工具）
├── client.py       # 小智 MCP 客户端
└── qt2_mcp.py      # QT2 量化系统代理

3.2 内置工具清单

3.3 MCP 服务器连接

# 客户端连接 MCP 服务器
import httpx

# 初始化
resp = await client.post("/mcp/message", json={
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "clientInfo": {"name": "my-app", "version": "1.0.0"}
    }
})

# 获取工具列表
resp = await client.post("/mcp/message", json={
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/list",
    "params": {}
})

# 调用工具
resp = await client.post("/mcp/message", json={
    "jsonrpc": "2.0",
    "id": 3,
    "method": "tools/call",
    "params": {
        "name": "ask_expert",
        "arguments": {
            "question": "劳动合同到期不续签需要赔偿吗？",
            "device_id": "device_001"
        }
    }
})

3.4 SSE 流式传输

MCP 支持 Server-Sent Events 实时推送：

# SSE 连接
async with httpx.AsyncClient() as client:
    async with client.stream("GET", "/mcp/sse") as resp:
        async for line in resp.aiter_lines():
            if line.startswith("data:"):
                event = json.loads(line[5:])
                # 处理事件

3.5 QT2 代理工具

MCP 服务器内置了 QT2 量化系统代理，支持远程调用量化分析：

3.6 自定义工具开发

在 src/mcp/tools.py 中添加新工具：

# 1. 定义工具 schema
MY_CUSTOM_TOOL = {
    "name": "my_custom_tool",
    "description": "工具描述...",
    "inputSchema": {
        "type": "object",
        "properties": {
            "param1": {"type": "string", "description": "参数说明"},
        },
        "required": ["param1"]
    }
}

# 2. 注册到工具列表
MISES_TOOLS.append(MY_CUSTOM_TOOL)

# 3. 在 server.py 中实现处理逻辑
# _handle_tools_call() 方法中添加:
elif tool_name == "my_custom_tool":
    return await self._call_my_custom(arguments)

4. 智能路由

4.1 概述

SmartRouter 根据查询复杂度、用户等级和系统状态，自动选择最优 LLM 模型。

代码位置：src/core/smart_router.py

4.2 路由因素

4.3 模型能力矩阵

MODEL_CAPABILITIES = {
    "anthropic/claude-sonnet-4":   {"reasoning": 9, "code": 9, "creative": 9, "speed": 7},
    "anthropic/claude-opus-4":     {"reasoning": 10, "code": 9, "creative": 9, "speed": 5},
    "anthropic/claude-haiku-4":    {"reasoning": 7, "code": 7, "creative": 7, "speed": 9},
    "openai/gpt-4.1":             {"reasoning": 9, "code": 9, "creative": 8, "speed": 7},
    "openai/gpt-4o-mini":         {"reasoning": 7, "code": 7, "creative": 7, "speed": 9},
    "openai/o1":                   {"reasoning": 10, "code": 9, "creative": 6, "speed": 3},
    # ... 更多模型
}

4.4 使用示例

from src.core.smart_router import SmartRouter, RoutingContext, UserTier

router = SmartRouter()
context = RoutingContext(
    user_id="user_001",
    user_tier=UserTier.PRO,
    query="请分析这段代码的时间复杂度",
    has_code=True,
)

decision = router.route(context)
print(decision.selected_model)       # "anthropic/claude-sonnet-4"
print(decision.complexity)           # QueryComplexity.CODE
print(decision.reason)               # "代码分析任务，选择高代码能力模型"
print(decision.estimated_cost_usd)   # 0.003
print(decision.fallback_models)      # ["openai/gpt-4.1", ...]

5. SCST 自进化训练

5.1 概述

SCST（Self-Critical Sequence Training）通过对比学习实现专家能力的持续进化。

代码位置：src/training/scst_trainer.py、src/training/scst_score.py

5.2 核心类

• SCSTrainer — 主训练器，实现 generate → score → reward → store 循环
• SCSTScoreEnsemble — 14 维多评分器集成
• SCSTScore — 标准化 [0,1] 分数
• SCSTReward — 奖励 = score(sample) - score(greedy)

5.3 训练流程

1. generate_pair()    → 对同一问题生成 sample 和 greedy 两个回答
2. score_pair()       → 用 14 维评分器打分
3. compute_reward()   → reward = score(sample) - score(greedy)
4. store()            → 正奖励的 sample 作为优质训练数据
5. finetune()         → 导出正样本用于模型微调

5.3 评分维度

5.4 API 接口

# 运行单步训练
POST /api/v1/level6/scst/train-step
{
  "question": "劳动合同解除的经济补偿标准？",
  "expert_id": "legal_expert_001"
}

# 批量训练
POST /api/v1/level6/scst/batch-train
{
  "questions": [...],
  "expert_id": "legal_expert_001",
  "batch_size": 10
}

# 导出训练数据
GET /api/v1/level6/scst/training-data?min_reward=0.1

6. 因果推理引擎

6.1 概述

基于图结构的因果关系推理，支持根因分析和影响链追踪。

代码位置：src/knowledge/graph/causal_query.py

6.2 边类型

6.3 使用示例

from src.knowledge.graph.causal_query import CausalQueryEngine

engine = CausalQueryEngine()

# 添加因果关系
engine.add_causal_relation("通货膨胀", "购买力下降", weight=0.8)
engine.add_causal_relation("购买力下降", "消费减少", weight=0.7)
engine.add_mitigation("通货膨胀", "央行加息")

# 追踪因果链
chain = engine.get_causal_chain("通货膨胀", max_depth=3, direction="forward")
# → ["通货膨胀", "购买力下降", "消费减少"]

# 根因分析
roots = engine.get_root_causes("消费减少")
# → ["通货膨胀"]

# 获取缓解措施
mitigations = engine.get_mitigations("通货膨胀")
# → ["央行加息"]

# 综合摘要
summary = engine.get_causal_summary("通货膨胀")

6.4 REST API

# 查询因果链
POST /api/v1/level6/causal/query
{"entity": "通货膨胀", "direction": "forward", "max_depth": 3}

# 综合摘要
GET /api/v1/level6/causal/summary?node_id=通货膨胀

# 添加因果关系
POST /api/v1/level6/causal/add
{"cause": "通货膨胀", "effect": "购买力下降", "weight": 0.8}

# 批量添加
POST /api/v1/level6/causal/batch-add
{"relations": [{"cause": "A", "effect": "B", "weight": 0.8}, ...]}

7. 离线 CoT 框架

7.1 概述

离线 Chain-of-Thought 框架对线上交互进行离线重评估，通过逐步推理发现错误并生成改进信号。

代码位置：src/training/offline_cot.py

7.2 核心类

• OfflineCoTTrainer — 主训练器，管理交互记录 → 采样 → CoT 重评估 → 奖励提取
• InteractionLog — 单次线上交互记录
• CoTStep — 单步推理
• CoTResult — CoT 输出（步骤 + 分数）
• CoTReward — 步骤级奖励（accuracy, coherence）
• CoTParser — 解析 CoT 标记（<step>、<verify>、<answer> 等）

7.3 工作流程

1. record_interaction()  → 记录线上交互（ai_bot.py 自动调用）
2. sample_for_cot()      → 按低分/复杂度优先采样
3. run_cot_batch()       → 批量 CoT 重评估
4. extract_rewards()     → 提取步骤级奖励信号
5. feed_to_scst()        → 回传奖励到 SCST 训练

7.3 CoT 标记语法

<step>第一步推理内容</step>
<step>第二步推理内容</step>
<verify>验证推理过程</verify>
<judgment>正确/错误/部分正确</judgment>
<overall_score>0.85</overall_score>
<weakest_step>2</weakest_step>
<answer>最终答案</answer>

7.4 调度

8. API 参考

8.1 Agent 编排 API

# 多智能体编排查询
POST /api/v1/level6/orchestrate
{
  "question": "分析新能源投资的法律风险和财务回报",
  "force_multi": false
}

# 获取编排统计
GET /api/v1/level6/orchestrator/stats

8.2 完整 Level 6 API 路由

9. 最佳实践

9.1 多智能体编排

9.2 MCP 工具开发

9.3 智能路由

📖 相关文档：Tool 使用指南 · API Reference · System Card