DojoAgents 主动式 Agent
源码级深度分析

📦 一、项目概览

一句话：DojoAgents 不是"回答问题"的金融 AI，而是自主完成全链路投研的 Agent 引擎——市场扫描、财报拆解、产业链分析、风险测算，Agent 自己判断"下一步该查什么、如何验证"。

实际源码目录结构

dojoagents/
├── agent/                    # ★ 核心 Agent 引擎 (41 个文件)
│   ├── loop.py               # ★ AgentLoop — 核心编排器 (1523行)
│   ├── runtime.py            # ★ Runtime — 全局依赖注入组装 (361行)
│   ├── harness.py            # ★ Harness 基类 — 自动验证框架
│   ├── harnesses/            # 3 个具体 Harness (portfolio/tool_orchestrated/artifact_synthesis)
│   ├── hooks/                # Agent 生命周期 Hook 系统
│   ├── guardrails.py         # 工具调用安全护栏
│   ├── compressor.py         # LLM 辅助上下文压缩
│   └── session_manager.py    # 会话持久化 (27KB)
├── tasks/                    # ★ 任务系统 — YAML 合约 + TASK.md 提示词
├── planning/                 # ★ 自动规划 — LLM 生成计划 → DAG 执行
├── multi_agent/              # ★ 多 Agent — 池化管理 + 事件驱动分发
├── cron/                     # ★ 定时调度 — APScheduler + Gateway 推送
├── memory/                   # 记忆系统 — Hook 自动同步
├── skills/                   # 技能系统 — 可复用工作流
├── plugins/                  # 插件系统 — PluginRegistry (34KB)
├── tools/                    # 工具系统 (24 个文件)
├── gateway/                  # 多通道聊天网关
├── dashboard/                # FastAPI + React 前端
└── utils/event_bus.py        # ★ 事件总线 (仅23行!) — 主动式 Agent 的神经中枢

技术栈

🏗 二、七层架构全景图

⚙ 三、主动式 Agent 的六大核心机制

DojoAgents 的"主动性"不是单一特性，而是由六层机制叠加而成的系统行为。

🔍 四、Agent 循环 — 核心引擎源码追踪

dojoagents/agent/loop.py 是整个系统最核心的文件，1523 行实现完整的 Agent 编排逻辑。

4.1 AgentLoop.run() 执行流程

4.2 关键代码：Agent 创建与执行

# loop.py:1134-1144 — 创建 Strands Agent 实例
agent = Agent(
    model=model,              # DojoStrandsModelBridge (桥接 LLM)
    messages=agent_messages,  # 历史会话
    tools=strands_tools,      # 桥接后的工具列表
    system_prompt=system,     # 动态构建的系统提示
    hooks=hooks,              # 5 个 Hook Provider
    plugins=plugins,          # 插件
    callback_handler=...,     # 流式输出回调
    agent_id=strands_agent_id,
    session_manager=strands_session_manager,
)

# loop.py:1170-1177 — 执行 Agent (多轮 Tool Calling)
result = await agent.invoke_async(
    prompt=user_prompt,
    invocation_state=invocation_state,
    limits=limits,  # Limits(turns=100) — 防止无限循环
)

4.3 关键代码：Guardrails — 多层工具护栏

# loop.py:830-995 — Before Tool Call Hook (简化版)
async def check_guardrails_before(event: BeforeToolCallEvent) -> None:
    tool_name = event.tool_use.get("name")
    args = event.tool_use.get("input") or {}

    # 1. 图片模式 → 阻止代码执行工具
    if tool_name in IMAGE_TURN_EXCLUDED_工具:
        event.cancel_tool = blocked_res["content"]

    # 2. 动态创建工具 (延迟绑定 — 支持运行时注册)
    if not event.selected_tool:
        event.selected_tool = DojoBridgedTool(...)

    # 3. execute_code 护栏 → LLM 辅助分类判断是否允许
    if tool_name in EXECUTE_CODE_TOOL_NAMES:
        classification = await classify_execute_code(code, ...)
        if blocked: event.cancel_tool = ...

    # 4. write_session_file 护栏 → LLM 判断写入安全性
    if tool_name == "write_session_file":
        classification = await classify_write_session_file(...)
        if blocked: event.cancel_tool = ...

    # 5. 通用护栏 → 规则引擎检测
    decision = self.guardrails.before_call(tool_name, args)
    if decision.should_halt:
        raise GuardrailHaltException(...)  # 彻底终止
    elif not decision.allows_execution:
        event.cancel_tool = ...  # 阻止此工具

    # 6. Harness 工具修复 → 任务模式下限制工具使用
    if active_harness:
        repaired = active_harness.repair_tool_calls(calls, state)
        blocked = active_harness.block_tool_call(call, state)

4.4 关键代码：After Hook — 事件驱动自动修复

# loop.py:1002-1045 — After Tool Call Hook (简化版)
async def check_guardrails_after(event: AfterToolCallEvent) -> None:
    # ★ 工具执行失败 → 发布事件，MultiAgentAutoDispatcher 自动修复
    if is_failed:
        failure_results = await event_bus.publish(
            "ToolExecutionFailed",
            {"tool_name": tool_name, "args": args, "error": raw_result}
        )
        if failure_results:
            event.result["content"] = [{"text": failure_results[0]}]
            # ↑ Reviewer Agent 的修复结果自动注入!

    # ★ 大数据量 (>5000字符) → 发布事件，Analyst Agent 自动分析
    if len(raw_result) > 5000:
        data_results = await event_bus.publish(
            "DataVolumeLarge",
            {"data_summary": raw_result[:2000] + "..." + raw_result[-1000:]}
        )
        if data_results:
            event.result["content"] = [{"text": data_results[0]}]
            # ↑ Analyst Agent 的洞察自动替换原始数据!

✅ 五、Harness 系统 — 自动验证与自我纠错

这是 DojoAgents 最独特的设计——Agent 不是一次性生成输出就结束，而是通过 Harness 系统持续自我验证，不满足条件就自动继续。

5.1 Harness 继承体系

TaskHarness (抽象基类)                    # agent/harness.py
├── matches(request, state) → bool        #   是否匹配当前任务
├── block_tool_call(call, state) → str    #   阻止不适合的工具调用
├── repair_tool_calls(calls, state) → []  #   修复/限制工具调用
├── validate_progress(state) → Decision   #   ★ 验证任务是否完成
└── build_recovery_prompt(dec, locale)    #   ★ 构建恢复引导

具体实现:
├── PortfolioTaskHarness        (21KB)    #   验证投资组合任务
├── ToolOrchestratedHarness              #   引导工具编排步骤
└── ArtifactSynthesisHarness             #   验证产物 Schema

5.2 核心逻辑：Harness Recovery Loop

# loop.py:1280-1330 — 自动验证与纠错循环
harness_recovery_turns = 0
while True:  # ★ 自动循环直到任务完成或达到上限!
    decision = active_harness.validate_progress(harness_state)
    if decision.complete:
        break  # 任务完成，退出循环

    # 构建恢复提示 (中英文双语)
    recovery_prompt = active_harness.build_recovery_prompt(decision, locale)

    # 达到上限 → 输出当前状态并停止
    if harness_recovery_turns >= max_recovery_turns:
        response_text = recovery_prompt
        break

    harness_recovery_turns += 1
    # ★ 自动重新调用 Agent，注入引导提示!
    result = await agent.invoke_async(
        prompt=recovery_prompt,
        invocation_state=invocation_state,
        limits=limits,
    )

5.3 ToolOrchestratedHarness — 步骤引导示例

# harnesses/tool_orchestrated.py:62-96
def validate_progress(self, state) -> HarnessDecision:
    # 1. 检查输出 Schema 验证 — 如果合法则完成
    filename, issues = self._output_write_issues(state)
    if filename and not issues:
        return HarnessDecision(complete=True)

    # 2. 引导下一步操作
    if not state.tool_results:
        issues.append("Start with get_market_overview.")
    elif not has_tool(state, "get_sector_movers"):
        issues.append("Call get_sector_movers before web_search.")
    elif not has_tool(state, "write_session_file"):
        issues.append("Finish by writing output with write_session_file.")

    # 3. 返回决策 — 允许继续但指定下一步
    return HarnessDecision(
        complete=False,
        issues=issues,
        next_steps=issues[:1],
        allow_extra_steps=True,  # ★ 关键: 允许继续
    )

⚡ 六、事件驱动自动化 — 23行的神经中枢

dojoagents/utils/event_bus.py 仅 23 行代码，却驱动了整个自主响应体系。

6.1 事件总线源码

# utils/event_bus.py — 仅 23 行!
class EventBus:
    def __init__(self):
        self._subscribers: dict[str, list[Callable]] = {}

    def subscribe(self, event_type: str, handler: Callable) -> None:
        self._subscribers.setdefault(event_type, []).append(handler)

    async def publish(self, event_type: str, payload: dict) -> list[Any]:
        results = []
        for handler in self._subscribers.get(event_type, []):
            res = await handler(payload)
            if res is not None:
                results.append(res)
        return results

event_bus = EventBus()  # 全局单例

6.2 事件流全景

🌊 七、完整数据流追踪

以下追踪一个典型主动式请求 "分析今天A股半导体板块" 经过的完整路径：

⚖ 八、主动式 vs 被动式 — 技术对比

💡 九、构建主动式 Agent 的六大洞察