nmfs-agents/DEVELOP.md

Rockchip Agents 开发指南

环境搭建

python3.10 -m venv venv
source venv/bin/activate
pip install -e ".[dev]"

认证

Agent 以子进程调用 claude --print，支持两种认证：

方式	配置	优先级
Claude CLI 登录	claude login，token 存储于 ~/.claude/	推荐，无需配置
API Key	设置 ANTHROPIC_API_KEY 环境变量	备选

架构概览

Agent → CLI 映射

DeveloperAgent.run(task)
  └─> subprocess.Popen([
        "claude", "--print",
        "--output-format", "json",
        "--no-session-persistence",
        "--model", "claude-sonnet-4-6",
        "--append-system-prompt", "<项目+模式上下文>",
        # auto/confirm 模式（confirm 依赖 claude-wx hook 级审批）：
        "--dangerously-skip-permissions",
        # report 模式：
        "--allowedTools", "Read,Glob,Grep",
        "<任务描述>"
      ],
      cwd="<项目路径>",
      env={CLAUDECODE 已清除},      # 允许在 Claude Code 会话中嵌套运行
  )

confirm 模式审批机制：通过 claude-wx hook 基础设施实现，每个 Write/Edit/Bash 工具调用前均发飞书卡片请求审批，而非任务级一次性审批。环境变量 CLAUDE_WX_AGENT_MODE=default 触发此行为。

任务队列状态机

pending → running → done
                 → failed
                 → waiting_approval → done（审批通过后 callback 触发）
                                    → failed（审批拒绝）
                                    → vetoed → pending（否决后重入队，带否决原因）

多队列架构

scheduler.py
  ├─> project_delivery 队列（tasks.db）    ─> 项目开发/研究类任务
  ├─> base_opt 队列（base_opt_tasks.db）   ─> 算法/硬件优化类任务
  └─> insight 队列（insight_tasks.db）     ─> 洞察/分析类任务

每个队列独立的 role_filter 确保任务只被有资格的 Agent 执行。

核心循环（scheduler → executor）

Scanner.scan_all()            # 扫描 README/TODO/CLAUDE.md
  └─> TaskQueue.enqueue()     # SQLite，BEGIN EXCLUSIVE 原子操作，去重
        └─> Executor.run_all_pending()  # asyncio Semaphore 限流并发
              ├─> _make_agent(role)     # 按 agent_role 路由到对应 Agent
              └─> FeishuNotifier.send_task_result()

常用命令

# 查看任务状态（多队列）
python scripts/run.py status

# 后台启动调度器
bash scripts/start_daemon.sh

# 监控日志
bash scripts/watch_daemon.sh

# 预览扫描结果（只打印，不入队）
python scripts/run.py scan

# 执行队列中下一个任务（须在普通终端执行）
python scripts/run.py run-next

# 全量测试
pytest tests/ -q

调试

查看任务队列

# project_delivery 队列
sqlite3 data/tasks.db "SELECT id,project,agent_role,title,status FROM tasks ORDER BY id DESC LIMIT 20;"

# base_opt 队列
sqlite3 data/base_opt_tasks.db "SELECT id,project,agent_role,title,status FROM tasks ORDER BY id DESC LIMIT 10;"

# insight 队列
sqlite3 data/insight_tasks.db "SELECT id,project,agent_role,title,status FROM tasks ORDER BY id DESC LIMIT 10;"

手动入队

source venv/bin/activate
python - <<'EOF'
from rockchip_agents.config import load_config
from rockchip_agents.core.queue import Task, TaskQueue
import asyncio, rockchip_agents.core.executor as ex

cfg = load_config()
q = TaskQueue()
q.enqueue(Task(
    project="embedding", type="code_review", title="分析代码质量",
    context="", priority=3, mode="report", agent_role="developer"
))
asyncio.run(ex.Executor(cfg, queue=q).run_next())
EOF

注意：claude --print 子进程需在 Claude Code 会话外 的普通终端执行。

Scanner 扫描来源

来源	提取内容	优先级
README.md 的 ## 已知问题	每一行问题 → fix_bug 任务	2
.py 文件中的 TODO/FIXME/HACK	注释 → code_review 任务	4
CLAUDE.md 含 ❌/⚠ 的行	约束违反 → fix_bug 任务	2
projects.yaml description 关键词	待修复/零检出 → fix_bug；FPS偏低 → improve_perf	2/3
兜底	每个项目固定生成定期代码巡检	5

自动跳过：venv, .venv, .git, __pycache__, node_modules, build, dist

架构优化版本管理

arch_optimize 任务入队
  └─> ArchitectAgent._optimize()
        ├─ ArchVersion.snapshot()          # 备份当前 .py 文件
        ├─ claude --dangerously-skip-permissions  # 应用架构改动
        ├─ pytest tests/ -q                # 验证测试
        ├─ 通过 → ArchVersion.promote()   # 晋升
        └─ 失败 → ArchVersion.rollback()  # 恢复备份

python scripts/run.py arch-versions yolo   # 查看版本历史
python scripts/run.py arch-restore yolo v001  # 手动恢复

添加新项目

在 configs/projects.yaml 添加：

projects:
  my-project:
    path: /data/company/my-project   # 省略则从 scan_dirs 自动发现
    mode: auto                        # auto | confirm | report
    description: "项目描述，含待修复/FPS偏低等关键词可自动生成任务"

添加新 Agent

1. 在 src/rockchip_agents/agents/ 新建 my_agent.py，继承 DeveloperAgent 或独立实现 run(task) -> AgentResult
2. 在 executor.py 的 _make_agent() 添加路由条目
3. 在 configs/agents.yaml 的对应队列 roles: 中添加新角色名
4. 在 tests/ 添加测试文件

配置说明

configs/projects.yaml

scan_dirs:               # 自动扫描目录（发现项目）
- /data/company
exclude_projects:        # 跳过的目录名（防止噪音）
- rknn-toolkit2
projects:                # 显式配置（可覆盖自动发现的项目）
  my-project:
    path: /data/company/my-project
    mode: confirm
    description: "项目描述"

configs/agents.yaml

scheduler:
  interval_hours: 1       # 扫描间隔
  max_concurrent: 3       # 全局最大并发
queues:
  project_delivery:
    db: data/tasks.db
    max_concurrent: 3
    scan_interval_hours: 1
    roles:                # 仅执行此列表中角色的任务
      - developer
      - architect

configs/devices.yaml

devices:
  rk3566:                 # 算法默认验证设备
    type: linux
    connect: ssh
    host: 192.168.123.183
    user: orangepi
    password: "orangepi"
    workspace: /home/orangepi/Desktop
  rk3588:                 # 高性能验证设备
    type: linux
    connect: ssh
    host: 192.168.123.181
    user: pi
    password: "123123"
    workspace: /home/pi/Desktop
    pre_cmd: "clashon && clashproxy on"

环境变量

变量	说明	必填
ANTHROPIC_API_KEY	Claude API 密钥（已 claude login 则不需要）	否
FEISHU_WEBHOOK_URL	飞书 Webhook，任务结果通知	否
FEISHU_BOT_TOKEN	飞书 Bot Token，confirm 模式 hook 审批需要	否
DB_PATH	任务数据库路径（默认 data/tasks.db）	否
DASHBOARD_PORT	Dashboard 端口（默认 8080）	否

项目记忆系统

data/project_memory.db 分三层：

层	读写规则
goals（目标）	仅 productizer/architect 可写
facts（事实）	任意 agent；architect 通过 [记忆] 标签写入
history（历史）	executor 每次任务完成/失败后自动写入

python scripts/run.py goals    # 查看项目长期目标
python scripts/run.py facts    # 查看项目知识事实
python scripts/run.py set-fact embedding api_port "8000"

Claude-WX 集成

Agent 模式	CLAUDE_WX_AGENT_MODE	飞书行为
auto	trust	工具执行通知，不阻塞
confirm	default	每个 Write/Edit/Bash 需飞书逐一审批
report	silent	静默，只读工具无需通知

配置 agents.yaml 中的 claude_wx_url 后，Executor 在任务开始/结束时调用 /agent/task-start 和 /agent/task-end 更新飞书调度中心。

Watchdog

core/watchdog.py 后台 daemon 线程（60s 间隔）：

• reset_stale_running(timeout=120min)：running 超时 → failed
• bump_stale_priorities(threshold=120min)：pending 久等 → priority-1（防饥饿）