1
# backend/packages/harness/deerflow/agents/lead_agent/agent.py - build_middlewares
2

3
def build_middlewares(config, model_name, agent_name=None, ...):
4
    middlewares = build_lead_runtime_middlewares(...)  # 基础运行时中间件
5

6
    # 1️⃣ 动态上下文注入（日期 + 记忆）
7
    middlewares.append(DynamicContextMiddleware(agent_name=agent_name))
8

9
    # 2️⃣ 技能激活（/skill-name 斜杠命令）
10
    middlewares.append(SkillActivationMiddleware(available_skills=available_skills))
11

12
    # 3️⃣ 技能工具策略（allowed-tools 过滤）
13
    middlewares.append(SkillToolPolicyMiddleware(available_skills=available_skills))
14

15
    # 4️⃣ 持久上下文（摘要 + 账本 + 技能注入）
16
    middlewares.append(DurableContextMiddleware(...))
17

18
    # 5️⃣ 上下文摘要压缩
19
    if summarization_middleware is not None:
20
        middlewares.append(summarization_middleware)
21

22
    # 6️⃣ 计划模式 TodoList
23
    if is_plan_mode:
24
        middlewares.append(TodoMiddleware(...))
25

26
    # 7️⃣ Token 用量追踪
27
    middlewares.append(TokenUsageMiddleware())
28

29
    # 8️⃣ 标题生成
30
    middlewares.append(TitleMiddleware())
31

32
    # 9️⃣ 记忆中间件
33
    middlewares.append(MemoryMiddleware(agent_name=agent_name))
34

35
    # 🔟 图像查看（仅视觉模型）
36
    if model_config.supports_vision:
37
        middlewares.append(ViewImageMiddleware())
38

39
    # 1️⃣1️⃣ MCP 路由提示
40
    if mcp_routing_middleware is not None:
41
        middlewares.append(mcp_routing_middleware)
42

43
    # 1️⃣2️⃣ 延迟工具过滤
44
    if deferred_setup.deferred_names:
45
        middlewares.append(DeferredToolFilterMiddleware(...))
46

47
    # 1️⃣3️⃣ SystemMessage 合并（兼容严格后端）
48
    middlewares.append(SystemMessageCoalescingMiddleware())
49

50
    # 1️⃣4️⃣ 子代理并发限制
51
    if subagent_enabled:
52
        middlewares.append(SubagentLimitMiddleware(max_concurrent=3, max_total=N))
53

54
    # 1️⃣5️⃣ 循环检测
55
    middlewares.append(LoopDetectionMiddleware.from_config(...))
56

57
    # 1️⃣6️⃣ Token 预算硬停
58
    middlewares.append(TokenBudgetMiddleware.from_config(...))
59

60
    # 1️⃣7️⃣ 自定义扩展中间件
61
    middlewares.extend(load_configured_extension_middlewares(...))
62

63
    # 1️⃣8️⃣ 终端响应保护
64
    middlewares.append(TerminalResponseMiddleware())
65

66
    # 1️⃣9️⃣ 安全终止检测
67
    middlewares.append(SafetyFinishReasonMiddleware.from_config(...))
68

69
    # 2️⃣0️⃣ 澄清拦截（必须最后）
70
    middlewares.append(ClarificationMiddleware())
71
    return middlewares

1
# backend/packages/harness/deerflow/subagents/executor.py - 隔离事件循环
2

3
# 🔑 全局持久化事件循环（避免每次执行创建/销毁）
4
_isolated_subagent_loop: asyncio.AbstractEventLoop | None = None
5
_isolated_subagent_loop_thread: threading.Thread | None = None
6

7
def _get_isolated_subagent_loop() -> asyncio.AbstractEventLoop:
8
    """返回用于隔离子代理执行的持久化事件循环"""
9
    global _isolated_subagent_loop, _isolated_subagent_loop_thread
10

11
    with _isolated_subagent_loop_lock:
12
        # 健康检查：线程存活 + 循环未关闭 + 循环运行中
13
        loop_is_usable = (
14
            _isolated_subagent_loop is not None
15
            and not _isolated_subagent_loop.is_closed()
16
            and _isolated_subagent_loop.is_running()
17
            and thread_is_alive
18
        )
19

20
        if not loop_is_usable:
21
            loop = asyncio.new_event_loop()
22
            started_event = threading.Event()
23
            thread = threading.Thread(
24
                target=_run_isolated_subagent_loop,
25
                args=(loop, started_event),
26
                name="subagent-persistent-loop",
27
                daemon=True,
28
            )
29
            thread.start()
30
            # 🔑 等待循环就绪（5秒超时）
31
            if not started_event.wait(timeout=5):
32
                raise RuntimeError("Timed out starting isolated subagent event loop")
33
            _isolated_subagent_loop = loop
34

35
        return _isolated_subagent_loop

1
# backend/packages/harness/deerflow/subagents/executor.py - SubagentResult
2

3
@dataclass
4
class SubagentResult:
5
    task_id: str
6
    trace_id: str
7
    status: SubagentStatus
8
    result: str | None = None
9
    error: str | None = None
10
    stop_reason: str | None = None  # 🔑 token_capped / turn_capped / loop_capped
11
    cancel_event: threading.Event = field(default_factory=threading.Event)
12
    _state_lock: threading.Lock = field(default_factory=threading.Lock)
13

14
    def try_set_terminal(self, status, *, result=None, error=None,
15
                         stop_reason=None, ...) -> bool:
16
        """设置终端状态 — 恰好一次（exactly-once 语义）"""
17
        if not status.is_terminal:
18
            raise ValueError(f"Status {status} is not terminal")
19

20
        with self._state_lock:
21
            # 🔑 第一个终端转换获胜；后续写入被忽略
22
            if self.status.is_terminal:
23
                return False
24

25
            self.result = result
26
            self.error = error
27
            self.stop_reason = stop_reason
28
            self.status = status
29
            return True

1
# backend/packages/harness/deerflow/skills/catalog.py - SkillCatalog
2

3
@dataclass(frozen=True)
4
class SkillCatalog:
5
    """不可变技能目录 — 纯搜索，无变更"""
6
    skills: tuple[Skill, ...]
7

8
    def search(self, query: str) -> list[Skill]:
9
        """三种查询形式（镜像 DeferredToolCatalog）"""
10

11
        # 1️⃣ 精确选择："select:data-analysis,deep-research"
12
        if query.startswith("select:"):
13
            wanted = {n.strip() for n in query[7:].split(",")}
14
            return [s for s in self.skills if s.name in wanted]
15

16
        # 2️⃣ 必选前缀："+podcast gen" → 名称必含 podcast，按 gen 排序
17
        if query.startswith("+"):
18
            parts = query[1:].split(None, 1)
19
            required = parts[0].lower()
20
            candidates = [s for s in self.skills if required in s.name.lower()]
21
            # ...排序逻辑
22
            return candidates[:MAX_RESULTS]
23

24
        # 3️⃣ 自由文本正则："chart visualization"
25
        regex = _compile_catalog_regex(query)
26
        scored = []
27
        for s in self.skills:
28
            searchable = f"{s.name} {s.description or ''}"
29
            if regex.search(searchable):
30
                # 🔑 名称匹配得分高于描述匹配
31
                scored.append((2 if regex.search(s.name) else 1, s))
32
        scored.sort(key=lambda x: x[0], reverse=True)
33
        return [s for _, s in scored][:MAX_RESULTS]

1
def _stable_tool_key(tool_call: dict) -> str:
2
    """生成稳定的工具调用标识 — 不同工具不同策略"""
3
    name = tool_call.get("name", "")
4
    args = tool_call.get("args", {})
5

6
    if name == "read_file":
7
        # 🔑 按 200 行桶分组 — 读同一区域视为相同
8
        start = args.get("start_line", 0)
9
        bucket = start // 200
10
        return f"read_file:{args.get('path')}:{bucket}"
11

12
    if name in ("write_file", "str_replace"):
13
        # 写操作：路径 + 内容哈希
14
        content_hash = hashlib.md5(str(args.get("content", "")).encode()).hexdigest()[:8]
15
        return f"{name}:{args.get('path')}:{content_hash}"
16

17
    # 默认：名称 + 完整参数哈希
18
    return f"{name}:{hashlib.md5(json.dumps(args, sort_keys=True).encode()).hexdigest()[:12]}"

1
# backend/packages/harness/deerflow/runtime/journal.py (883 行)
2

3
class RunJournal:
4
    """LangChain callback → RunEvent 标准化审计"""
5

6
    # 🔑 作为 LangChain CallbackHandler 挂载
7
    # 自动捕获：
8
    #   - LLM 调用开始/结束 + token 用量
9
    #   - 工具调用开始/结束 + 结果
10
    #   - 中间件事件（guardrail 拦截、循环检测等）
11
    #   - 错误和异常
12

13
    # 🔑 累积 token 用量（按模型分组）
14
    # token_usage_by_model: {"claude-sonnet": {"input": 12000, "output": 3000}}
15

16
    # 🔑 批量写入 RunEventStore（避免 per-step I/O）
17
    async def flush(self):
18
        await self._event_store.put_batch(self._buffer)

1
# backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py (962 行)
2

3
class _ProcessWideLimiter:
4
    """跨事件循环的 LLM 调用并发限制器"""
5

6
    # 🔑 为什么不用 asyncio.Semaphore？
7
    # Semaphore 绑定到第一个使用它的事件循环
8
    # Lead Agent 在主循环，子代理在隔离循环
9
    # → 必须用 threading 原语实现跨循环限制
10

11
    # 🔑 关键设计：
12
    # - Cap 不可变：启动时解析一次，永不修改
13
    # - Lossless waiter handoff：取消时许可不丢失
14
    # - call_soon_threadsafe 跨循环唤醒
15

16
# 智能重试分类：
17
_BUSY_PATTERNS = ("server busy", "服务繁忙", "稍后重试", ...)
18
_QUOTA_PATTERNS = ("insufficient_quota", "余额不足", ...)
19
_BURST_PATTERNS = ("limit_burst_rate", "请求速率增长过快", ...)
20

21
# 🔑 Per-exception 重试预算：
22
_RETRY_BUDGET_OVERRIDES = {
23
    "StreamChunkTimeoutError": 2,  # 已等待 120s，只重试 1 次
24
}