作者： 人生导师
日期： 2026 年 6 月
状态： 设计阶段，持续迭代

这篇文章记录了一个面向 Android 逆向工程的 Multi-Agent 协作系统设计。不是教程，是一个正在落地的系统架构。

动机很简单：逆向工程中大量工作是重复性的模式识别和数据处理，但关键判断点需要人的经验。如果能让 AI 承担 70% 的重复劳动，人只在关键节点做决策，效率会有质的提升。

现有的 AI 编程工具（Claude Code、Cursor、Hermes）都不是为逆向场景设计的。它们缺乏逆向专用的工具集成、没有针对逆向工程的校验机制、不支持跨天持久化的分析状态。所以需要自己搞一个。

主控集权 + 插件化 + 上下文隔离 + 文档驱动协作

为什么这样设计：

Google 的 A2A 协议太早期，生态几乎没有。用它等于自己实现大部分东西，协议本身反而成了额外负担。不如自己设计一套轻量的。

四种消息，职责清晰：

现实场景：trace agent 分析过程中发现一个 native 调用，给 IDA agent 发 REQUEST："帮我查 sub_5200 的交叉引用，结果写到 knowledge/facts/ 里"。IDA agent 如果手头有 TASK 在跑就排队，空闲就直接干。不需要所有协作都绕主控一圈。

如果对方拒绝或超时呢？发起方先尝试降级处理（比如跳过这个信息继续分析），降级也搞不定再上报主控重新调度。

只有要你干活的消息进上下文，通知和状态不进。

这保证每个 agent 的对话历史是纯净的——只有任务指令 + 自己的工具调用历史。共享知识全在文件系统里，需要的时候自己去读，不会被推送进来污染思考过程。

最开始考虑过让 agent 之间通过消息直接传递分析结果，但这有两个问题：

所以选了文档驱动：agent 产出信息写到共享目录，别的 agent 需要时自己去读。消息总线只负责通知"有新东西了"。

逆向工程中的信息有明确的成熟度层级：

agent 发现的现象，未经验证，不含推测。

已通过工具或代码确认的信息，可复现、可验证。

基于观察和事实形成的推测，明确标记为未验证。

通过三层校验确认的信息，可作为后续分析的可信前提。

这样既能避免上下文爆炸，也能保留逆向过程中每一步的证据链。事后复盘时，可以从 Conclusion 一路追溯到原始 Observation。

模型会编造结论，但我们有办法拦住。

硬校验是最可靠的一层——用代码直接验证结论中引用的具体数据。

硬校验的局限：只能验证具体值（地址、常量、行号），验不了逻辑推理。所以需要第二层。

一个结论的可信度跟硬校验覆盖率直接相关：

不是所有分析任务都需要全部 agent 同时在线。从"固定编队"变成"按需组装"：

每个 agent 是一个独立插件，带自己的代码、prompt、工具和校验规则。

manifest 只是声明，真正干活的是 agent.py。每个 agent 有自己的业务逻辑和编排策略：消息怎么接、工具怎么调、下一步怎么决定、报告怎么格式化。这些用纯配置搞不定。

为防止职责漂移（agent 用久了变成什么都干的缝合怪），manifest 中的 subscribes 是硬约束——超出声明范围的消息类型直接拒收。需要新能力就改 manifest 并审核，不是悄悄在代码里加。

逆向一个目标可能跨好几天，进度不能丢。

每个 agent 启动时：

不需要记住对话历史，状态全在文件里。这也是为什么选文档驱动——agent 随时可以冷启动恢复工作。

用一个典型场景走一遍完整协作流程：

统一接口切模型，按任务特性选模型：

agent 数量少（4-5 个），WS 连接管理简单，不需要 Redis 或消息队列。

每个 agent 挂自己的 MCP server：

这个系统的核心难点不在代码量——可能总共就几千行 Python。难的是：

这些都是可以逐步解决的工程问题，不是原理性障碍。

最坏情况：系统跑不起来，但积累的 prompt 模板、校验方法、工具集成经验单独拿出来也有用。

最好情况：系统自动完成 70% 的分析工作，人只在关键节点做决策，逆向效率翻几倍。

GitHub链接是：102K9s2c8@1M7s2y4Q4x3@1q4Q4x3V1k6Q4x3V1k6Y4K9i4c8Z5N6h3u0Q4x3X3g2U0L8$3#2Q4x3V1k6V1K9Y4y4C8L8X3y4^5L8g2)9J5c8V1c8#2j5$3E0m8k6$3g2F1N6l9`.`.

┌─────────────────────────────────────────────────────┐
│                    消息总线                          │
│            SQLite + FastAPI + WebSocket              │
└──────┬────────┬────────┬────────┬────────┬──────────┘
       │        │        │        │        │
  ┌────▼────┐┌──▼─────┐┌──▼────┐┌──▼────┐┌──▼──┐
  │  主模型  ││IDA+Jadx ││Trace   ││Unidbg  ││  我  │
  │ 调度+搜索││静态分析  ││算法分析││补环境   ││Leader│
  └─────────┘└────────┘└───────┘└───────┘└─────┘
       │        │        │        │
  ┌───▼────┐┌──▼────┐┌──▼───┐┌──▼───┐
  │搜索工具 ││IDA MCP ││trace  ││unidbg │
  │网络判断 ││Jadx CLI││数据   ││Java   │
  └────────┘└───────┘└──────┘└──────┘

              ┌──────────────────┐
              │   knowledge/     │
              │  (共享文件系统)   │
              └──────────────────┘

from enum import Enum

class MessageType(Enum):
    TASK = "task"          # 主控派发，必须执行，不可拒绝
    REQUEST = "request"    # 同级协作请求，可拒绝/排队，优先级低于 TASK
    REPORT = "report"      # 任务完成报告，通知相关方"我产出了新东西"
    STATUS = "status"      # 状态广播（idle/busy/blocked），用于协作调度

{
    "id": "msg_001",
    "from": "trace_agent",
    "to": "ida_agent",           # null = 广播
    "type": "request",           # task / request / report / status
    "content": "查 sub_5200 的交叉引用，结果写到 knowledge/facts/",
    "priority": "normal",        # high / normal / low
    "task_id": "task_003",       # 关联的任务 ID
    "timeout": 60,               # 秒
    "timestamp": "2026-06-02T10:00:00Z"
}

1. 收到消息，看 type
2. TASK → 进入上下文，立即开始干活
3. REQUEST → 看自己状态：
   - 空闲 → 接受，进入上下文，开始干
   - 忙碌 → 排队 or 拒绝
4. REPORT → 不进上下文，只记录"有新文档可读"
5. STATUS → 更新对方状态表（用于发 REQUEST 时判断对方是否空闲）
6. 产出结论 → 写文档 + 发 REPORT 通知

Observation（观察）→ Fact（事实）→ Hypothesis（假设）→ Conclusion（结论）

{
    "type": "observation",
    "agent": "trace_agent",
    "content": "sub_5200 存在 64 次重复状态更新",
    "source": "trace_line_47_to_128",
    "timestamp": "2026-06-02T10:05:00Z"
}

{
    "type": "fact",
    "agent": "ida_agent",
    "content": "sub_12A0 调用了 sub_5200（偏移 0x12B4 处 BL 指令）",
    "evidence": ["ida_xref_sub_12A0_0x12B4"],
    "timestamp": "2026-06-02T10:03:00Z"
}

{
    "type": "hypothesis",
    "agent": "trace_agent",
    "content": "sub_5200 可能实现了 SHA256（64 轮 + 常量 0x6A09E667 匹配）",
    "supporting_evidence": ["observation_001", "fact_003"],
    "confidence": "medium",
    "next_step": "用 unidbg 跑一遍对比标准 SHA256 输出",
    "timestamp": "2026-06-02T10:10:00Z"
}

{
    "type": "conclusion",
    "agent": "trace_agent",
    "content": "sub_5200 确认为标准 SHA256 实现",
    "verified_by": ["hard_check_h0_h7", "validator_agent", "unidbg_output_match"],
    "evidence_chain": ["observation_001", "fact_003", "hypothesis_001"],
    "timestamp": "2026-06-02T10:30:00Z"
}

project_xxx/
├── status.json              # 全局进度快照
├── knowledge/
│   ├── observations/        # 未验证的现象
│   ├── facts/               # 工具确认的事实
│   ├── hypotheses/          # 待验证的推测
│   └── conclusions/         # 通过校验的结论
├── messages/                # 消息历史（总线持久化）
├── traces/                  # trace 原始数据
├── unidbg/                  # unidbg 项目代码
└── failed_attempts/         # 失败尝试（防止重蹈覆辙）

工作 agent 输出结论
    ↓
┌────────────────────────────────────────┐
│ 第一层：自校验                         │
│  - 硬校验：代码验证具体值/常量/行号    │
│  - 软校验：模型自查推理链完整性        │
│  - 过了 → 第二层                       │
│  - 没过 → 打回重做（最多 3 次）        │
└───────────────┬────────────────────────┘
                ↓
┌────────────────────────────────────────┐
│ 第二层：校验 agent（无状态冷启动）     │
│  - 每次清空上下文，零先入为主          │
│  - 输入：结论 + 证据 + 原始数据        │
│  - 独立判断有没有矛盾                  │
│  - PASS → 结论入库（写 conclusions/）  │
│  - FAIL → 打回重做                     │
│  - UNCERTAIN → 上报第三层              │
└───────────────┬────────────────────────┘
                ↓
┌────────────────────────────────────────┐
│ 第三层：我                             │
│  - 只看 UNCERTAIN 和反复 FAIL 的       │
│  - 做路径决策和最终判断                │
└────────────────────────────────────────┘

def hard_verify(conclusion: dict, trace_data: list) -> dict:
    """验证结论中引用的具体值是否在原始数据中存在"""
    results = {"passed": [], "failed": [], "coverage": 0.0}
    
    for assertion in conclusion["assertions"]:
        if assertion["type"] == "register_value":
            line = trace_data[assertion["line_number"]]
            if assertion["register"] in line and \
               line[assertion["register"]] == assertion["value"]:
                results["passed"].append(assertion)
            else:
                results["failed"].append(assertion)
                
        elif assertion["type"] == "constant":
            found = any(assertion["value"] in line for line in trace_data)
            if found:
                results["passed"].append(assertion)
            else:
                results["failed"].append(assertion)
    
    total = len(results["passed"]) + len(results["failed"])
    results["coverage"] = len(results["passed"]) / total if total > 0 else 0
    return results

你是一个独立的校验员。你不知道之前发生了什么。
以下是一个分析结论和支撑证据。

【结论】
{conclusion}

【原始证据】
{evidence}

请验证：
1. 结论中引用的具体数据（地址、值、行号）是否在证据中存在且一致
2. 推理链是否有跳跃（A→B→C，每一步是否有依据）
3. 是否存在证据中的信息与结论矛盾
4. 结论是否遗漏了证据中的关键信息

输出格式：
- 判定：PASS / FAIL / UNCERTAIN
- 理由：具体说明哪里有问题或为什么通过
- 建议：如果 FAIL，建议从哪个方向重新分析

plugins/trace_agent/
├── manifest.yaml      # 元数据：名称、版本、订阅声明
├── system.md          # 系统 prompt
├── agent.py           # 核心逻辑：消息处理、工具编排、状态管理
├── tools/             # 专属工具实现
│   ├── frida_hook.py
│   └── log_parser.py
└── validators/        # 校验规则
    └── trace_check.py

name: trace_agent
description: "执行 trace 分析，输出函数调用序列和数据流"
version: 0.1.0

model: claude-sonnet-4
prompt: system.md

tools:
  - frida_trace
  - addr2line
  - log_parser

subscribes:
  - task
  - request

publishes:
  - report
  - status

def load_plugin(name: str):
    manifest = read_manifest(f"plugins/{name}/manifest.yaml")
    agent_module = import_module(f"plugins.{name}.agent")
    agent = agent_module.create(manifest)
    agent.connect_to_bus(bus_url)
    registry[name] = agent

def unload_plugin(name: str):
    agent = registry.pop(name)
    agent.disconnect()
    agent.shutdown()

{
    "target": "libxxx.so",
    "goal": "还原 sign 签名算法",
    "started": "2026-06-01",
    "phase": "algorithm_identification",
    "progress": {
        "entry_located": true,
        "call_chain_mapped": true,
        "algorithm_identified": false,
        "implementation_done": false,
        "verified": false
    },
    "confirmed_facts": [
        "入口函数: sub_12A0 (Java_com_xxx_Sign_generate)",
        "调用链: sub_12A0 → sub_3F00 (key生成) → sub_5200 (加密核心)"
    ],
    "blocked_paths": [
        "sub_7000-sub_8000 是 VMP 保护，静态分析无效"
    ],
    "next_steps": [
        "trace agent 分析 sub_5200 的执行流"
    ]
}

目标：还原 libxxx.so 中的 sign 签名算法

1. 我 → 主控（TASK）：分析 libxxx.so 的签名生成逻辑

2. 主控：
   - 搜索网上有没有现成分析文章
   - 判断：没有直接可用的，需要自己分析
   - → ida_agent（TASK）：定位签名入口函数，从 JNI_OnLoad 开始找
   - 广播（STATUS）：任务启动，目标是 sign 算法还原

3. IDA agent：
   - 通过 IDA MCP 找到 JNI 注册表
   - 定位到 native 函数 Java_com_xxx_Sign_generate → sub_12A0
   - 分析调用链：sub_12A0 → sub_3F00 → sub_5200
   - 写 knowledge/facts/entry_sub_12A0.json
   - 写 knowledge/facts/call_chain.json
   - → trace_agent（REQUEST）：入口确认 sub_12A0，帮忙收集执行 trace
   - 广播（REPORT）：入口定位完成，详情见 knowledge/facts/

4. Trace agent：
   - 收到 REQUEST，当前空闲，接受
   - 收集 sub_12A0 的执行 trace
   - 发现 64 轮循环 + 常量 0x6A09E667
   - 写 knowledge/observations/64_rounds.json
   - 写 knowledge/hypotheses/sub_5200_sha256.json
   - 自校验：trace 中搜索 SHA256 的 H0-H7 初始值，全部命中 ✓（硬校验覆盖率 87%）
   - 校验 agent 验证：PASS
   - 升级为 conclusion，写 knowledge/conclusions/sub_5200_sha256.json
   - → unidbg_agent（REQUEST）：算法确认 HMAC-SHA256，帮忙补环境验证
   - 广播（REPORT）：算法识别完成，详情见 conclusions/

5. Unidbg agent：
   - 收到 REQUEST，接受
   - 读 knowledge/conclusions/ 了解算法信息
   - 搭建 unidbg 环境，加载 libxxx.so
   - 补 JNI 环境（GetStringUTFChars 等）
   - 调用 sub_12A0，输入已知参数
   - 输出与抓包结果对比：匹配 ✓
   - 写 knowledge/conclusions/verification_pass.json
   - 广播（REPORT）：验证通过，算法还原正确

6. 主控：
   - 收到各 REPORT
   - 读取 conclusions/ 整合最终报告
   - → 我（REPORT）：分析完成，sign 算法为 HMAC-SHA256，
     key = sub_3F00(timestamp + device_id)，完整实现见 unidbg/

import litellm

# trace agent：需要精确推理
response = litellm.completion(
    model="claude-sonnet-4-20250514",
    messages=[...]
)

# unidbg agent：代码生成为主，用便宜的
response = litellm.completion(
    model="deepseek/deepseek-chat",
    messages=[...]
)

CREATE TABLE messages (
    id INTEGER PRIMARY KEY,
    from_agent TEXT NOT NULL,
    to_agent TEXT,                -- NULL = 广播
    type TEXT NOT NULL,           -- task / request / report / status
    content TEXT,
    priority TEXT DEFAULT 'normal',
    task_id TEXT,
    timeout INTEGER,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    acknowledged INTEGER DEFAULT 0
);

我（CLI 输入）+ 1 个 trace agent + 消息总线
- 验证消息收发机制
- 验证 REQUEST 的接受/拒绝流程
- 验证知识文档的读写
预计：1-2 周

在 Phase 1 基础上加入：
- 硬校验代码
- 校验 agent（无状态冷启动）
- 三层校验的完整流程
- 校验覆盖率统计
预计：1 周

加入其他 agent：
- 主模型（调度 + 搜索）
- IDA + Jadx agent
- Unidbg agent
- 持久化和恢复机制
预计：2-3 周

拿真实目标跑：
- 先跑简单的（非 VMP、标准算法）
- 再跑中等难度（魔改算法）
- 最后挑战 VMP 保护的
根据实战持续调整
预计：持续

作者： 人生导师
日期： 2026 年 6 月
状态： 设计阶段，持续迭代

• 主控（主模型/人）是唯一能下达强制命令的角色
• 各 agent 是独立插件，按需加载卸载，按场景组装
• 各 agent 有独立上下文，只有任务指令进入对话历史
• agent 之间可以互相发协作请求，像同事之间请求帮忙
• 跨 agent 的知识共享走文件系统，不走上下文注入
• 人是 leader，终审 + 路径决策

• TASK：命令。只有主控能发，agent 必须执行。超时未响应上报主控
• REQUEST：请求帮忙。任何 agent 都能发，对方可以拒绝或排队。超时未响应发起方自己降级处理
• REPORT：通知。干完活了告诉大家"有新产出"，具体内容在文档里
• STATUS：心跳。状态变化时广播（idle ↔ busy ↔ blocked），用于 REQUEST 的调度决策

• TASK 只有主控能发
• REQUEST / REPORT / STATUS 任何 agent 都能发
• agent 不能通过 REQUEST 强制对方干活——对方有权拒绝
• REQUEST 优先级永远低于 TASK，不能通过 priority=high 覆盖

• agent 产出信息，按成熟度写入对应目录
• Observation 可以升级为 Fact（通过工具验证后）
• 多个 Fact 可以支撑一个 Hypothesis
• Hypothesis 通过三层校验后升级为 Conclusion
• 只有 Conclusion 可以作为后续分析的可信前提
• Hypothesis 可以用来指导探索方向，但不能作为论据

• 无状态：每次任务完成后清空上下文，下次冷启动
• 不知道历史：不知道之前谁说了什么，避免 confirmation bias
• 只看证据：给它原始数据和结论，让它独立判断
• 输出格式固定：PASS / FAIL / UNCERTAIN + 理由

[培训]《冰与火的战歌：Windows内核攻防实战》！从零到实战，融合AI与Windows内核攻防全技术栈，打造具备自动化能力的内核开发高手。