文档中心

ReconAgent 技术文档

SQL 对账 Agent 的完整技术参考，涵盖架构设计、权限管控、RAG 检索、记忆系统、Subagent 协同、自进化机制与可观测性评测体系。

系统架构

LangGraph DAG、并行执行、FastAPI 服务层设计

权限管控

三层防护体系、AST 检查、表级字段级白名单

RAG 检索

Hybrid BM25 + 向量检索、RRF 融合、降级策略

记忆系统

三层记忆架构、情节提炼、语义规则持久化

Subagent 体系

SQL生成/反思/Skill审查三专职Agent协同

开发者指引

仓库结构、本地环境搭建、数据库设计速览

自进化机制

错误学习、记忆健康、Skill 持续优化

可观测性与评测

Golden Set、4 维 Metric、TraceLogger、反馈闭环

ReconAgent 基于 LangGraph 构建，所有组件可独立替换。文档持续更新，反映最新架构决策。

入门

开发者指引

面向初次接触 ReconAgent 的技术同学，帮你在 30 分钟内看懂仓库结构、核心模块与数据库设计。

仓库目录结构

整个项目按「层」组织：Agent 逻辑在 recon_core/，HTTP 服务在 apps/，数据与索引在 data/。

ReconCore-SQL-Reconciliation-Agent/
├── recon_core/          # 核心框架层
│   ├── agents/            # Agent 定义（ReconAgent 主体在此）
│   ├── tools/             # 工具集（SQLTool / RAGRetriever 等）
│   │   └── builtin/       # 内置工具实现
│   ├── context/           # GraphState 与上下文管理
│   ├── core/              # LangGraph DAG 编排核心
│   ├── skills/            # Skill 加载与生命周期
│   └── observability/     # 可观测性（链路追踪、日志）
│
├── apps/
│   └── ui/                # 前端页面（纯 HTML / JS）
│       ├── index.html     # 控制台主界面
│       ├── docs.html      # 本文档页
│       └── landing.html   # 产品介绍落地页
│
├── data/                  # 数据层
│   ├── enterprise_mock.db # 企业级模拟库（36张表，9000+行）
│   ├── production_mock.db # 生产级模拟库（6张表，带8种差异）
│   ├── eval_data.sqlite   # 评测数据集
│   ├── schema_index*.json # Schema RAG 索引
│   └── generate_*.py      # 数据生成脚本
│
├── knowledge_base/        # RAG 知识库（Markdown 文档）
├── memory_store/          # 记忆持久化（SQLite）
├── skill_library/         # Subagent / Skill 定义（.md）
├── recon_cases/           # 历史对账案例（few-shot）
└── tests/                 # 单元 / 集成测试

核心模块关系

用户自然语言问题 │ ▼ ┌─────────────────────────────────────────────────┐ │ FastAPI (apps/ui → apps/api) │ │ POST /query / GET /sessions │ └───────────────────┬─────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────┐ │ ReconAgent (recon_core/agents/) │ │ │ │ plan ─► parallel_act ─► observe ─► reflect ─► end│ │ │ │ │ │ │ Schema SQLTool RangeGuard │ │ Linker (RAG辅助) 错误反馈闭环 │ └─────────────────────────────────────────────────┘ │ ┌───────────┼───────────┐ ▼ ▼ ▼ SQLite DB BM25 Index Memory Store (data/*.db) (schema_index) (memory_store/)

数据库说明：两套模拟库

项目提供两套 SQLite 模拟数据库，用途不同，按需选择：

数据库文件	规模	设计目的	使用建议
production_mock.db	6 表 / ~1000 行	植入 8 种差异（金额差、缺失、重复、超期等），验证对账能力	快速跑通 Agent 对账流程
enterprise_mock.db	36 表 / 9000+ 行	覆盖用户、商品、营销、财务等多业务噪音，测试 Schema Linking	评估 Agent 在真实生产噪音下的表定位能力

两套库均由 data/generate_*.py 脚本生成，可随时重建。每次生成会打印差异分布，方便核查。

免责声明 — 本项目所有数据表（含字段名、表名、业务逻辑）均为模拟数据，系参考同类开源项目的通用数据模型或由 AI 自动生成，不代表任何真实公司、平台或产品的实际数据结构。数据内容（主播名称、金额、订单号等）全部为随机生成的虚构信息，不涉及任何真实业务数据，仅供学习与演示使用。

enterprise_mock.db 业务域速览

业务域	表（数量）	典型表名
直播电商核心	6	live_sessions / live_gmv / order_amount / settlements / refunds / commissions
用户体系	4	users / user_profiles / user_tags / user_login_logs
商品中心	4	products / product_categories / product_inventory / product_price_history
营销活动	4	campaigns / ad_spend / campaign_budgets / coupon_records
供应链	4	suppliers / purchase_orders / warehouses / logistics_records
财务中心	4	finance_bills / bank_statements / tax_records / cost_center_allocation
客服系统	3	complaints / complaint_followups / satisfaction_surveys
平台运营	3	anchor_contracts / anchor_performance / platform_rules
风控	2	risk_alerts / blacklist
系统审计	2	operation_logs / data_sync_tasks

对账差异类型速查

production_mock.db 和 enterprise_mock.db 均在直播电商核心表中植入了 8 种差异，Agent 对账时可验证这些问题是否被发现：

差异代号	类型	描述	涉及表
D1	金额差异	order 金额与 GMV 偏差 > 5%	live_gmv ↔ order_amount
D2	数据缺失	GMV 有记录，订单表无对应数据	live_gmv ↔ order_amount
D3	幽灵订单	session_id = NULL 的孤立订单	order_amount
D4	重复订单	同一 session 有 2 条 order 记录	order_amount
D5	超期结算	settle_date 超过 expected_date 超 8 天	settlements
D6	结算金额错	settle_amount ≠ gmv − platform_fee	settlements
D7	退款未扣除	有退款记录但结算金额未减去退款	settlements ↔ refunds
D8	分佣比例异常	commission_rate > 30% GMV	commissions

工具层：SQLTool 的三个子工具

recon_core/tools/builtin/sql_tool.py 是 Agent 与数据库交互的唯一入口，对外展开为三个子工具：

子工具名	作用	典型调用场景
sql_schema	查询指定表的字段列表、类型和前 3 行示例数据	Agent 做 Schema Linking 时主动探查表结构
sql_execute	执行 SELECT 语句，返回最多 50 行 Markdown 表格	执行对账 SQL 获取结果
sql_validate	用 SQLite EXPLAIN 做语法校验，不实际执行	执行前预检 SQL 合法性

所有子工具均拒绝 DROP / DELETE / UPDATE / INSERT / ALTER / CREATE，仅允许只读操作。

接入新数据库（配置方式）

切换 Agent 使用的数据库只需两步：

# 1. 实例化 SQLTool 时指向目标数据库
from recon_core.tools.builtin.sql_tool import SQLTool

tool = SQLTool(db_path="data/enterprise_mock.db")
registry.register_tool(tool)  # 自动展开为 sql_schema / sql_execute / sql_validate

# 2. 更新 schema_index 让 RAG 检索到正确表
# 将 schema_index_enterprise.json 的路径写入 RAGRetriever 配置即可

重新生成数据库后记得同步更新 schema_index.json，否则 Agent Schema Linking 会查到旧索引。

典型对账问题示例

以下是向 Agent 提问的参考问题，可用于验证各差异类型是否被正确检测：

D1 / D2 验证

「帮我找出直播 GMV 和订单金额差异超过 5% 的场次」

D3 / D4 验证

「哪些订单的 session_id 为空，或同一场次有重复订单？」

D5 / D6 验证

「列出结算超期或结算金额计算错误的场次」

D7 验证

「哪些场次有退款记录但结算金额没有扣除？」

D8 验证

「找出分佣比例超过 30% 的异常场次」

跨域挑战

「找出被风控标记过的主播，其 GMV 是否存在异常？」

架构

系统架构

基于 LangGraph DAG 的对账推理流程，每个节点纯函数、状态显式传递，让复杂多步推理可观测、可调试、可回滚。

整体分层

flowchart TB
  UI["用户 / 控制台 UI"]
  API["FastAPI  HTTP / SSE 层"]
  DAG["LangGraph — ReconAgent DAG\nplan → parallel_act → observe → reflect → end"]
  TOOL["工具层：SQL Executor · Schema Linker · RAG"]
  STORE["存储层：SQLite · BM25 Index · Memory Store"]
  UI --> API --> DAG --> TOOL --> STORE
  style UI    fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style API   fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style DAG   fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style TOOL  fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style STORE fill:#f7f9fc,stroke:#91a5c6,color:#24324a

LangGraph DAG 节点

节点	职责	关键输入 / 输出
plan	将自然语言拆解为并行 SQL 子任务	user_query → parallel_plan
parallel_act	asyncio.gather 并发执行多条 SQL	parallel_plan → parallel_results
observe	汇总结果、Range Guard 合理性校验	parallel_results → observation
reflect	发现异常时生成修正建议	observation → reflection
end	格式化最终答案返回	reflection → final_answer

并行执行机制

对账场景往往需要同时查询多张表或多个数据库，并发执行将延迟从 O(N) 降至 O(1)。

async def parallel_act(state: GraphState) -> GraphState:
    tasks = [execute_sql(step.sql, step.db) for step in state.parallel_plan]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    ...

数据流示意

flowchart LR
  Q["用户输入\n比较 A/B 表 6 月差异"] --> P["plan 节点"]
  P --> A["查询 table_A 6月"]
  P --> B["查询 table_B 6月"]
  A --> OB["observe 节点\n差异检测 + Range Guard"]
  B --> OB
  OB -- 有异常 --> RF["reflect 节点"]
  OB -- 无异常 --> END["end\n生成对账报告"]
  RF --> END
  style Q fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style P fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style OB fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style RF fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style END fill:#edf8f3,stroke:#2f8c6a,color:#24324a
  style A fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style B fill:#f7f9fc,stroke:#91a5c6,color:#24324a

部署结构

Docker Compose
├── recon-agent   (FastAPI + LangGraph)
└── nginx         (反向代理 + 静态文件)

# 快速部署
./deploy.sh ui   # 秒级前端热更
./deploy.sh all  # 全量重建

架构决策记录（ADR）

以下是项目关键架构决策的摘要。每条决策均包含选型背景与被否决的替代方案。

编号	决策	选型理由	被否决的替代方案
ADR-001	选择 LangGraph 作为 Agent 编排框架	状态显式传递，每步可观测、可回滚；相比 AutoGen 更适合有确定 DAG 拓扑的对账场景	AutoGen（隐式状态难调试）、纯 LangChain Chain（不支持条件分支）
ADR-002	采用 BM25 + 向量 Hybrid RAG	BM25 处理精确表名/字段名匹配，向量检索覆盖语义同义词；单用任一方均有盲区	纯向量 RAG（精确词命中率低）、纯 BM25（语义泛化弱）
ADR-003	三层记忆：Working / Episodic / Semantic	对账场景有明确的短/中/长期记忆需求；Episodic 提供 few-shot 案例，Semantic 持久化规则	单一 SQLite 表存所有记忆（无法按生命周期管理）
ADR-004	三层 SQL 安全防护（意图 + AST + 白名单）	字符串黑名单易被绕过；AST 检查语法树层面拦截；白名单限制爆炸半径	仅正则过滤 DDL 关键词（容易被注释或大小写绕过）
ADR-005	使用 SQLGlot 做静态 AST 安全检查	支持多方言、可扩展、纯 Python 无额外依赖；EXPLAIN 只捕运行时错误，无法识别危险语义	sqlite3 EXPLAIN（仅校验语法，不识别危险操作类型）

安全

SQL 权限管控

三层防护体系，从意图过滤到运行时白名单，防止 LLM 生成危险 SQL，保障数据安全。

三层防护体系

flowchart TD
  NL["用户自然语言"] --> L1["Layer 1：意图过滤\n关键词黑名单、意图分类"]
  L1 --> L2["Layer 2：AST 静态安全检查\nSQLGlot 解析，禁止 DDL/DML"]
  L2 --> L3["Layer 3：运行时权限过滤\n表级 / 字段级白名单"]
  L3 --> SQL["执行 SQL"]
  style NL  fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style L1  fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style L2  fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style L3  fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style SQL fill:#edf8f3,stroke:#2f8c6a,color:#24324a

Layer 2：SQLGlot AST 检查

只允许 SELECT，DDL/DML 在语法树层面直接拦截，不依赖字符串匹配，防止绕过。

import sqlglot

FORBIDDEN = {
    sqlglot.exp.Drop, sqlglot.exp.Delete,
    sqlglot.exp.Insert, sqlglot.exp.Update,
    sqlglot.exp.Create, sqlglot.exp.AlterTable,
}

def check_sql_safety(sql: str) -> bool:
    tree = sqlglot.parse_one(sql)
    for node in tree.walk():
        if type(node) in FORBIDDEN:
            raise PermissionError(f"禁止操作: {type(node).__name__}")
    return True

Layer 3：白名单配置

roles:
  analyst:
    allow_tables: [orders, payments]
    deny_columns: [user_phone, id_card_number]
  admin:
    allow_tables: "*"
    deny_columns: []

错误分类与重试策略

错误类型	处理方式	重试
PermissionError	立即终止，返回拒绝提示	否
SyntaxError	将错误上下文反馈给 LLM 重生成	最多 3 次
TimeoutError	终止并提示查询过于复杂	否
DataError	反馈给 observe 节点合理性判断	否

自我修正闭环

flowchart TD
  FAIL["SQL 执行失败"] --> OBS["observe 节点\nlast_sql_error 捕获\nobs_count += 1"]
  OBS --> ROUTE{obs_count < MAX_RETRY?}
  ROUTE -- 是 --> RF["reflect 节点\n携带错误重新生成 SQL"]
  ROUTE -- 否 --> FALLBACK["降级：返回'无法完成，请简化问题'"]
  RF --> FAIL
  style FAIL     fill:#fff2f2,stroke:#c75252,color:#24324a
  style OBS      fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style ROUTE    fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style RF       fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style FALLBACK fill:#f7f9fc,stroke:#91a5c6,color:#7185a5

权限错误不进入重试循环，避免 LLM 反复尝试绕过限制。

检索

RAG 检索增强

Hybrid RAG：BM25 精确匹配 + 向量语义检索，RRF 融合排序，将业务口径定义注入 prompt 减少幻觉。

两种检索方式对比

特性	BM25（关键词）	向量（语义）
精确词匹配	极强	弱
同义词/近义词	弱	强
速度	极快 ms 级	较慢
部署复杂度	低（纯 Python）	高（需向量库）

Hybrid 融合架构

flowchart LR
  Q["用户问题"] --> BM["BM25 检索"]
  Q --> VEC["向量检索"]
  BM --> RRF["RRF 融合排序\nscore = Σ 1/(k+rank)  k=60"]
  VEC --> RRF
  RRF --> TOP["Top-K 文档"]
  TOP --> PROMPT["注入 Prompt → LLM"]
  style Q      fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style BM     fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style VEC    fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style RRF    fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style TOP    fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style PROMPT fill:#edf8f3,stroke:#2f8c6a,color:#24324a

知识库组织

knowledge_base/
├── schema/          # 表结构描述、字段注释
│   ├── orders.md
│   └── payments.md
├── business/        # 业务口径定义
│   ├── metrics.md   # GMV、DAU 等指标
│   └── rules.md     # 结算规则、对账规则
└── examples/        # 历史对账 SQL 示例（few-shot）
    └── golden_set.jsonl

RAG vs Memory 对比

维度	RAG	Memory
存储内容	静态业务文档	动态会话记录
更新频率	低（手动更新）	高（每次对话）
适合场景	口径定义、schema 描述	用户偏好、历史错误

两者协同：RAG 提供"知识"，Memory 提供"经验"。

评估指标

以下数据来自项目内置评测集 data/eval_data.sqlite（共 120 条人工标注问答对），在 enterprise_mock.db 上运行，测试时间 2025-06。数字仅供参考，实际效果受 LLM 版本、数据规模影响。

指标	实测值（BM25 降级模式）	目标（Hybrid 模式）	说明
Recall@5	82%	>85%	Schema Linking Top-5 命中率
MRR	0.71	>0.75	Mean Reciprocal Rank，越高表示相关文档排名越靠前
SQL 执行准确率（有 RAG）	78%	>80%	对照 eval_data 正确答案，SQL 输出结果一致
SQL 执行准确率（无 RAG）	61%	—	关闭 RAG 的对照组，体现 RAG 增益 +17%

记忆

记忆系统

三层记忆架构让 Agent 积累经验，从历史案例中学习，避免重复犯错。

三层记忆模型

flowchart TB
  WM["Working Memory\n当前会话 GraphState\n会话结束自动清除"]
  EM["Episodic Memory\n历史对账案例\n按 session_id 索引，可检索回放"]
  SM["Semantic Memory\n提炼的业务规则与用户偏好\n长期有效，置信度衰减后遗忘"]
  WM -->|会话结束写入| EM
  EM -->|相同模式出现 N 次| SM
  SM -->|置信度衰减/被反驳| FORGET["遗忘"]
  style WM     fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style EM     fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style SM     fill:#edf8f3,stroke:#2f8c6a,color:#24324a
  style FORGET fill:#f7f9fc,stroke:#91a5c6,color:#7185a5

Working Memory

GraphState 就是 Working Memory，随节点传递，自动随 LangGraph 生命周期管理。

class GraphState(TypedDict):
    user_query: str
    parallel_plan: list[SQLStep]
    parallel_results: list[QueryResult]
    observation: Observation
    reflection: str
    last_sql_error: str | None
    obs_count: int
    final_answer: str

持久化表设计对比

两张核心表承载不同粒度的记忆，共同存放在 memory_store/ 下：

-- Episodic Memory：原始会话快照，粒度"一次对话"
CREATE TABLE episodic_case (
    id          TEXT PRIMARY KEY,   -- UUID
    session_id  TEXT,                   -- 关联前端会话
    user_query  TEXT,                   -- 原始自然语言问题
    sql_used    TEXT,                   -- 最终执行的 SQL（可能经过自我修正）
    anomaly     BOOLEAN,                -- 是否检测到异常
    error_type  TEXT,                   -- 若执行失败：SyntaxError / PermissionError 等
    created_at  DATETIME
);

-- Semantic Memory：提炼后的规则，粒度"一条知识"
CREATE TABLE semantic_rule (
    id          TEXT PRIMARY KEY,   -- UUID
    rule_text   TEXT,                   -- 自然语言规则，注入 Prompt 时直接引用
    source_ids  TEXT,                   -- 关联的 episodic_case.id（JSON 数组）
    confidence  REAL DEFAULT 1.0,       -- 0.0~1.0，每日衰减 0.05
    auto        BOOLEAN DEFAULT TRUE,   -- TRUE = 自动提炼，FALSE = 人工写入
    refuted     BOOLEAN DEFAULT FALSE,  -- 被用户显式纠正后置 TRUE，停止使用
    created_at  DATETIME,
    updated_at  DATETIME
);

维度	episodic_case	semantic_rule
粒度	一次对话	一条提炼规则
写入时机	每次会话结束自动写入	同类失败 ≥ 3 次触发提炼
生命周期	永久保留（可按 session 清除）	confidence 衰减至 0.2 后删除
Prompt 注入	few-shot 示例（相似案例 Top-K）	直接拼入规则说明段落
回滚方式	删除对应 session 行	设置 refuted=TRUE 或一键清空 auto=TRUE

记忆提升规则

记忆在三层之间按以下规则流转：

1. Working → Episodic：会话正常结束后，本次执行的 SQL、用户问题、是否发现异常等字段写入 episodic_case 表。

2. Episodic → Semantic：同一失败模式在 Episodic 中出现 ≥ 3 次时，触发规则提炼，写入 semantic_rule 表，带 confidence 字段。

3. 遗忘机制：Semantic 规则的 confidence 每日衰减 0.05；若被用户显式纠正，confidence 直接清零并标记为 refuted。低于 0.4 的规则停止注入 Prompt，低于 0.2 的规则物理删除。

持久化存储

memory_store/
├── episodic.db   # 情节案例
├── semantic.db   # 提炼规则
└── audit.db      # 执行审计日志

全部使用 SQLite，零外部依赖，部署简单，数据文件可直接备份。

协同

Subagent 体系

专职 Agent 协同：SQL 生成 / 反思 / Skill 审查，职责隔离、相互校验。

角色分工

flowchart TB
  MAIN["ReconAgent 主流程\nplan → parallel_act → observe → reflect → end"]
  MAIN -->|"委托"| SG["SQL Generator\n将子任务转化为可执行 SQL"]
  MAIN -->|"委托"| RA["Reflect Agent\n分析异常根因，提出修正"]
  MAIN -->|"定期触发"| SR["Skill Reviewer\n审查 skill_library 有效性"]
  style MAIN fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style SG   fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style RA   fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style SR   fill:#f7f9fc,stroke:#91a5c6,color:#24324a

三个 Subagent 详细职责对比

Subagent	触发时机	输入	输出	异常处理
SQL Generator	plan 节点为每个并行子任务委托	子任务描述 + Schema RAG Top-K + Semantic 规则	sql / confidence / tables_used / reasoning	confidence < 0.6 → 返回 DONE_WITH_CONCERNS，主流程记录告警
Reflect Agent	observe 节点检测到异常或 SQL 执行失败	上一次 SQL + 错误信息 + observation 结果	修正后的 SQL 或根因分析文本	重试 ≥ 3 次 → 返回 BLOCKED，人工介入
Skill Reviewer	每日定时任务 / 手动触发	skill_library/ 下全量 .md 文件列表	每个 Skill 的审查报告（维度评分 + 建议）	检测到危险操作 → 立即标记 QUARANTINE，暂停使用

SQL Generator — 真实 I/O 示例

以"找出 GMV 与订单金额差异超 5% 的场次"为例：

// 输入（由主流程构造）
{
  "task": "查询 live_gmv 与 order_amount 差异超过 5% 的直播场次",
  "schema_context": "live_gmv(session_id, gmv_amount, date)\norder_amount(session_id, total_amount, date)",
  "semantic_rules": ["差值计算使用 ABS 避免方向影响", "GMV 与订单比对需 LEFT JOIN 保留无订单的场次"]
}

// 输出（SQL Generator 返回）
{
  "sql": "SELECT g.session_id, g.gmv_amount, o.total_amount,\n       ABS(g.gmv_amount - o.total_amount) / g.gmv_amount AS diff_ratio\nFROM live_gmv g LEFT JOIN order_amount o USING (session_id)\nWHERE ABS(g.gmv_amount - COALESCE(o.total_amount,0)) / g.gmv_amount > 0.05",
  "confidence": 0.91,
  "tables_used": ["live_gmv", "order_amount"],
  "reasoning": "使用 LEFT JOIN 保留无订单记录；COALESCE 将 NULL 视为 0 避免除零"
}

Subagent 通信协议

class SubagentResponse(BaseModel):
    status: Literal["DONE", "DONE_WITH_CONCERNS", 
                    "BLOCKED", "NEEDS_CONTEXT"]
    output: dict
    concerns: list[str] = []         # DONE_WITH_CONCERNS 时填写
    context_needed: list[str] = []   # NEEDS_CONTEXT 时填写所需信息

状态	含义	主流程响应
DONE	正常完成	继续执行下一节点
DONE_WITH_CONCERNS	完成但有风险（如 confidence 低）	继续执行，但记录告警写入 audit.db
BLOCKED	无法完成（超出重试上限）	暂停流程，返回用户"无法完成，请简化问题"
NEEDS_CONTEXT	缺少必要信息	向用户追问 context_needed 中列出的字段

Skill Reviewer 审查维度

维度	检查项	不通过处理
安全性	是否包含危险操作（文件删除、网络请求）	立即 QUARANTINE，暂停使用
有效性	依赖的 API / 数据源是否仍可访问	标记 STALE，下次触发前人工确认
质量	输出格式是否规范、是否有测试用例	生成改进建议，不自动禁用
时效性	上次更新时间，是否需要同步最新口径	超过 30 天未更新提示 OUTDATED

扩展新 Subagent

---
name: my-subagent
description: 做什么用途，何时触发
---

[System Prompt 内容]

在 skill_library/ 下新建 .md 文件，主 Agent 启动时自动扫描，无需修改代码。

进化

自进化机制

Agent 在运行中自动识别失败模式、提炼经验规则、更新内部知识，不重训模型持续提升准确率。

进化闭环

flowchart TD
  Q["用户提问"] --> EXEC["Agent 执行"]
  EXEC --> EVAL["结果评估\n正确 / 错误 / 用户反馈"]
  EVAL -- 正确 --> EP["写入 Episodic Memory\n成功案例沉淀"]
  EVAL -- 错误 --> AN["错误分析\n聚类失败模式"]
  AN --> SM["提炼规则\n写入 Semantic Memory"]
  SM --> NEXT["下次类似问题\n规则自动注入 Prompt"]
  style Q    fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style EXEC fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style EVAL fill:#eef3ff,stroke:#4f6fd9,color:#24324a
  style EP   fill:#edf8f3,stroke:#2f8c6a,color:#24324a
  style AN   fill:#fff2f2,stroke:#c75252,color:#24324a
  style SM   fill:#edf8f3,stroke:#2f8c6a,color:#24324a
  style NEXT fill:#edf8f3,stroke:#2f8c6a,color:#24324a

四个进化维度

SQL 生成质量

连续失败时聚类分析错误模式，提炼为语义规则

Schema Linking

记录表名/字段名错误映射，持续修正词汇表

Skill 进化

Skill Reviewer 定期标记过时 Skill，建议更新或删除

记忆健康度

清理低置信度规则，防止记忆"污染"

触发时机

触发条件	进化动作
会话结束	成功案例写入 Episodic Memory
SQL 连续失败 3 次	提炼错误模式为语义规则
用户显式纠正	高权重规则立即写入
每日定时任务	Memory Hygiene + Skill Review
手动触发	全量优化（控制台 → 设置 → 功能）

进化边界与防护

规则置信度低于 0.7 不生效（仅记录）；涉及权限的变更需人工确认；所有自动写入的规则标记 auto=true，可一键回滚。进化不修改模型权重，仅修改 prompt 上下文。

评测 · 观测

可观测性与评测体系

从 Golden Set 离线评测到线上 TraceLogger 追踪，再到用户反馈闭环，让 Agent 表现完全可量化、可追溯、可迭代。

整体架构

flowchart TB
  subgraph EVAL["评测层 — tests/eval/"]
    GS["Golden Set\n50+ cases"]
    RUNNER["runner.py\n批量执行"]
    EA["Exec-Accuracy\n执行结果比对"]
    SEM["Semantic-Match\n语义等价判断"]
    GS --> RUNNER --> EA & SEM
  end
  subgraph OB["可观测层 — TraceLogger"]
    TRACE["JSONL + HTML\n双格式审计轨迹"]
    RG["Range Guard\n数据合理性校验"]
  end
  subgraph FB["反馈闭环"]
    FEED["POST /feedback\nthumbsup / thumbsdown"]
    EVO["SkillReviewer\n异步分析根因"]
    SM2["semantic.json\n规则提炼写入"]
    FEED --> EVO --> SM2
  end
  RUNNER --> TRACE
  OB --> FB
  SM2 --> RUNNER
  style EVAL fill:#eef3ff,stroke:#4f6fd9
  style OB fill:#edf8f3,stroke:#2f8c6a
  style FB fill:#f2efff,stroke:#725ad6

Golden Set 与四维指标

基准测试覆盖 50 个案例，横跨 6 类意图，每条记录结构如下：

// tests/eval/golden_set.jsonl — 每行一条 case
{
  "id": "mj-007",
  "query": "支付和订单金额不一致的记录",
  "intent_label": "multi_table_join",
  "difficulty": "hard",
  "expected_sql": "SELECT o.id, o.amount, p.amount FROM orders o JOIN payments p ...",
  "expected_result_summary": "订单/支付金额不一致",
  "tags": ["diff"]
}

指标	含义	计算方式
Exec-Accuracy	SQL 执行结果与参考一致	result_set hash 比对（含 6 种宽松策略）
Semantic-Match	自然语言答案语义等价	关键词覆盖 / LLM-as-Judge（可注入）
Avg Latency	平均执行延迟 (ms)	含 LLM + SQL 执行全链路
P95 Latency	尾部延迟	第 95 百分位，衡量稳定性

宽松匹配策略按优先级：精确 hash → 两者均空 → 截取前 N 列 → 列顺序不同 → 主键集合一致 → 单值数值误差 < 0.1%。避免等价 SQL 被误判为错误。

运行方式与质量门禁

# 全量 50 条评测
python -m tests.eval.runner --target v2 --db data/eval_data.sqlite

# smoke test：只跑前 10 条
python -m tests.eval.runner --target v2 --limit 10

# v1 vs v2 对比
python -m tests.eval.runner --target v2 --compare v1

# EA < 75% 时 CI 返回非零退出码，阻断部署
if agg["exec_accuracy"] < 0.75:
    sys.exit(1)

TraceLogger 事件类型

每次 Agent 执行自动生成 trace-{session_id}.jsonl 和 trace-{session_id}.html。

事件	触发时机	关键字段
`session_start / end`	会话开始 / 结束	agent_name, duration, total_tokens
`llm_request / response`	LLM 调用前后	model, messages, usage.total_tokens
`tool_call / result`	SQL 执行前后	sql, db_path, rows_count, duration_ms
`range_guard_triggered`	合理性校验拦截	check_type, value, action
`circuit_breaker`	熔断器触发	tool_name, state

Range Guard — 数据合理性校验

在 observe.py 节点内置四类断言，拦截"SQL 语法正确但业务语义异常"的结果：

金额负值检测

总金额 < 0 → 告警用户，可能是 JOIN 方向错误

比率超 100%

退款率 / 转化率 > 1.0 → 分母分子可能混淆

COUNT 爆炸

COUNT > 百万 → 通常是笛卡尔积 JOIN 未加条件

对账空结果

对账查询返回空集 → 时间窗口设置可能有误

线上取 Query 的三条路径

路径	数据源	特点	适用场景
路径 1	sessions.sqlite messages 字段	全量，当下可用	普通问题采样
路径 2	episodic.json (outcome=0)	带质量标签，差评 case	最高价值改进样本
路径 3	query_log 表（建议落地）	全量 + status 字段，可按错误筛选	系统性 Golden Set 扩充

# 从 episodic.json 导出差评 case（可直接追加到 golden_set.jsonl）
python - << 'EOF'
import json, pathlib
cases = json.loads(pathlib.Path("memory_store/episodic.json").read_text())
bad = [c for c in cases if c.get("outcome", 1) == 0]
for c in bad:
    print(json.dumps({
        "id": f"online-{c['trace_id'][:8]}",
        "query": c["query"],
        "intent_label": c.get("intent", "unknown"),
        "difficulty": "hard",
        "expected_sql": c.get("sql", ""),  # 需人工修正
        "expected_result_summary": "待标注",
        "tags": ["online_fail"]
    }))
EOF

反馈闭环

flowchart LR
  FB["用户点 ❌\nPOST /feedback"] --> EP["写入 episodic.json\noutcome=0, user_flag=1"]
  FB --> REV["异步 SkillReviewer\n分析失败根因"]
  REV --> SM["提炼语义规则\nsemantic.json"]
  SM --> PLAN["下次 plan 节点\n自动注入规则"]
  style FB fill:#fff2f2,stroke:#c75252,color:#24324a
  style EP fill:#f7f9fc,stroke:#91a5c6,color:#24324a
  style REV fill:#f2efff,stroke:#725ad6,color:#24324a
  style SM fill:#edf8f3,stroke:#2f8c6a,color:#24324a
  style PLAN fill:#edf8f3,stroke:#2f8c6a,color:#24324a

反馈接口返回 importance 分（0~1），越高代表该 case 对进化越有价值；SkillReviewer 异步运行，不阻塞 API 响应。

评测迭代 SOP

1. 运行全量 Golden Set（50 条）
   python -m tests.eval.runner --target v2

2. 查看意图维度报告（哪类意图 EA 最低）
   重点关注 exec_accuracy < 70% 的意图类型

3. 收集失败案例，按类型分组
   语法错误 / 幻觉表名 / 语义偏差 / REJECT 误判

4. 修复最高频失败类型（每轮只改一类）
   经验值：每轮修复一类高频错误，通常提升 10~15 个百分点

5. 回归全量，验证无副作用

6. 从线上差评取 5~10 条补充 Golden Set
   python scripts/export_bad_cases.py >> tests/eval/golden_set.jsonl

7. 重复直到 EA ≥ 目标值（当前基线 84%）