Blade Agent — if/else 分支分析报告

Python 代码分支密度分析与重构建议

生成时间：2026-06-30 · 统计范围：core/src, host/src, server/src（排除 solutions, skills, builtin）

5,746
if/elif/else 总数
2,690
Guard Clause（47%）
3,056
可优化分支（53%）
76,772
总代码行数

分支模式分布

模块分支分布

Host 子模块热力图

484
sessions
440
orchestrator
412
tool_host
362
sandbox
229
projections
192
memory
174
llm
136
execution
114
tokenize
51
其它

Top 30 高密度文件

# 文件 if/else 密度 风险

重构建议（按优先级排序）

P0 · 收益最大

projections/builder.py — 95 个分支

-30~45
预估减少

on_event() 是最大分支源：事件类型分发 + active state 分发 + pause 子流程混在一起

dict dispatch 状态表 提取子函数
具体方案
  • 拆成 PRE_TURN_EVENT_HANDLERS 和 ACTIVE_TURN_EVENT_HANDLERS 两个 dict
  • compaction 流程用状态表替代嵌套 if
  • 提取 _handle_child_pause() 和 _handle_chat_end_pause()
  • 有投影快照测试覆盖，改动风险低
P0 · 接近行数上限

socketio_handlers.py — 156 个分支

-35~50
预估减少

chat_send() 和 _run() 过大；token 鉴权、what-if、replay、共享广播、异常回滚全混在一起

策略表 拆模块 提取公共广播
具体方案
  • 抽 authenticate_socket_token() 策略表处理 sk-blade-v3/legacy/cookie
  • 抽 emit_to_owner_and_viewers() 统一广播
  • 抽 rollback_whatif_if_applied()
  • GIS/ASR/logs handler 拆到独立模块
  • trigger_mode 用 dict 模板
P0 · 接近行数上限

routes/agent_board.py — 118 个分支

-18~28
预估减少

1453 行接近 1500 上限；create_mode、assignee_type 是明确策略分发

dict dispatch 拆文件 提取公共逻辑
具体方案
  • CREATE_PROJECT_HANDLERS = {"blank": ..., "local": ..., "gitea": ...}
  • 抽 resolve_agent_assignment() 同时服务 create/update
  • 按 project/task/git 路由拆文件
P1 · 需谨慎

runtime/loop.py — 148 个分支

-25~40
预估减少

_execute_tool_call() 同时做解析、hook、审批、执行、追踪、pause/terminal 判断

pipeline 拆分 默认对象
具体方案
  • 拆成 _prepare_tool_call → _run_tool_with_hooks → _finish_from_tool_result
  • 重复 llm_config is not None 建 request_meta 默认对象
  • ⚠️ 不要动 JSON scanner 状态机
P1

autotune.py — 114 个分支

-20~30
预估减少

on_event() 典型事件分发；chat:end status chain；coverage 提取分支散

EVENT_HANDLERS dict 提取 Collector
P1

routes/sessions.py — 109 个分支

-20~35
预估减少

Guard clause 多且合理，复杂点在 create_session()、upload_files()、MIME 判断

提取 validator context manager handler 分发
P2 · 低风险

llm/client.py — 93 个分支

-20~32
预估减少

模型兼容策略分支（reasoning replay、tool result image mode、stream delta 累积）

策略 dict StreamAccumulator
P2 · 低风险

tool_host/_document_readers.py — 91 个分支

-18~30
预估减少

文件类型分发和解析器选择适合注册表；guard 是正常格式校验

EXTENSION_KIND_MAP LIGHT_READERS 注册表

全局性重构策略（砍掉一半需要系统性手段）

仅 Top 8 文件可减少 ~185-290 个分支，但目标砍半需要减少 ~2,873 个

01

事件分发 → dict dispatch / match-case

autotune/builder/bridge/projections 等十几个文件都有 on_event() 模式。定义统一的 EventDispatcher 基类或装饰器，一次性消除所有事件分发 if/elif chain

影响面最广
02

None 检查 → 默认值 / 构造时保证

1,171 个 is None / is not None 检查。相当一部分可以在 dataclass/pydantic model 构造时给默认值，或用 field(default_factory=...) 消除

1,171 处
03

字符串比较 → Enum + dict 分发

455 个 == "string"。大量是 event kind / status / type 硬编码字符串，换成 StrEnum 后可用 dict 或 match-case 集中处理

455 处
04

isinstance → 多态 / singledispatch

409 个 isinstance 检查。特别集中在 projections 和 loop 中，可用方法分派或 @singledispatch 替代

409 处
05

重复 guard → 提取 validator / 装饰器

routes 层大量重复的权限/参数校验模式（检查 session 所有权、检查 user_id、检查文件存在性），可用 FastAPI dependencies 或装饰器统一

routes 层集中

第二轮分析 — Guard Clause 能否通过结构优化消除？

Pydantic 验证 · FastAPI Depends · Optional 收窄 · 代码实例

2,690
Guard Clause 总数
180~260
可真正消除
120~170
高投入产出比目标
~2,430
应保留（合法边界防御）

Guard Clause 分类

消除手段收益预估

Guard Clause 详细分类

类型 数量 典型模式 能否消除
Falsy Check 1,143 if not X: return/raise 部分可消 Pydantic 非空约束 + 默认值
None Check 414 if X is None: raise/return 部分可消 Depends + Optional 收窄
Equality Check 332 if X == Y: return value 大部分保留 业务逻辑判断
Other 289 if blocking: return ... 保留 控制流必要分支
Membership Check 133 if X in {a, b}: return 保留 集合过滤
Not-None Check 129 if X is not None: return X 部分可消 缓存命中/可选服务
Type Check 122 if isinstance(X, str): return 可重写 match-case / singledispatch
Raise Guard 81 if bad: raise ValueError 保留 契约式编程
Length Check 40 if len(X) > N: return 保留 边界校验
Dict .get() 7 if raw.get(key): return 可消 默认值

五种消除手段（含代码示例）

P0 · 最高收益

FastAPI Depends 合并重复 404/403 Guard

-70~95
预估消除

359 个 raise HTTPException guard 中有 265 个是 None/not check。software_factory.py 单文件重复 8 次 "if record is None: raise HTTPException(404)"

代码示例
BEFORE — 8 处重复
# software_factory.py (L251, L423, L447, L719...)
record = store.get_for_user(software_id, user_id=user_id)
if record is None:
    raise HTTPException(404, "Software project not found")
return _to_response(record)
AFTER — 1 个 Depends
def get_software_project_or_404(
    software_id: int,
    user_id: str = Depends(get_current_user_id),
) -> SoftwareFactoryRecord:
    record = store.get_for_user(software_id, user_id=user_id)
    if record is None:
        raise HTTPException(404, "Software project not found")
    return record

@router.get("/software/{software_id}")
def get_software(
    record=Depends(get_software_project_or_404),
):
    return _to_response(record)  # guard 消失

项目已有 _get_session_or_404 和 _get_project_or_404 雏形，但未统一成 Depends。推广到 sessions、skills、scenarios 路由可消 70-95 个。

P0 · 低难度

Pydantic Field / Literal / StringConstraints

-45~60
预估消除

路由层大量手动 strip() + if not: raise 400，以及枚举值手动校验，可以在 Pydantic model 层自动完成

代码示例
BEFORE — 手动校验
# factory_v2.py L509
title = req.title.strip()
if not title:
    raise HTTPException(400, "任务标题不能为空")
if req.assignee_type not in {"user", "agent"}:
    raise HTTPException(400,
        "assignee_type 仅支持 user 或 agent")
AFTER — Pydantic 声明式
NonEmptyStr = Annotated[str,
    StringConstraints(
        strip_whitespace=True, min_length=1
    )]

class CreateBoardTaskRequest(BaseModel):
    title: NonEmptyStr
    assignee_type: Literal["user", "agent"] = "user"

# 路由函数中直接使用，guard 消失
# Pydantic 自动返回 422 + 详细字段错误

注意：错误码从手写 400 变成 FastAPI 默认 422，前端若依赖具体状态码或文案需同步调整。

P1 · 中高难度

Optional 收窄 / 类型精确化

-25~45
预估消除

414 个 is None + 129 个 is not None 中，部分是"上游已验证但类型没表达"。通过更精确的类型标注在上游保证非 None

代码示例
BEFORE — 下游反复判断
# SessionInfo.user_id: str | None = None
# 导致下游所有 owned 路径反复判断

session = self._index.get(session_id)
if session is None or not session.user_id:
    return None
return self._user_home_paths.get(
    session.user_id
)
AFTER — 收窄类型
# 引入已验证的子类型
@dataclass(frozen=True, slots=True)
class OwnedSessionInfo(SessionInfo):
    user_id: str  # 非 Optional，构造时保证

def require_owned_session(
    session_id: str,
) -> OwnedSessionInfo | None:
    """验证一次，下游不再需要 None check"""
    ...

# 下游直接用，guard 消失
session = self.require_owned_session(sid)
return self._user_home_paths.get(
    session.user_id
)

风险：容易碰到测试 fixture、本地开发模式、旧会话和可选服务的 edge case。不建议全局改 SessionInfo，而是在 owned 路径引入收窄 helper。

P2 · 低风险

isinstance → match-case / singledispatch

-30~45
预估消除

122 个类型检查 guard 中，适合替换的是内部多形态分发（JSON 边界的类型检查应保留）

代码示例
BEFORE
# solutions/manifest.py L312
if isinstance(value, str):
    return os.path.expandvars(value)
if isinstance(value, list):
    return [_expand_env(i) for i in value]
if isinstance(value, dict):
    return {str(k): _expand_env(v)
            for k, v in value.items()}
AFTER — singledispatch
@singledispatch
def _expand_env(value):
    return value

@_expand_env.register(str)
def _(v): return os.path.expandvars(v)

@_expand_env.register(list)
def _(v): return [_expand_env(i) for i in v]

@_expand_env.register(dict)
def _(v): return {str(k): _expand_env(x)
                  for k, x in v.items()}
P2 · 谨慎操作

Early Return 简化

-20~30
预估消除

host 层 331 个 if not X: return。部分单层条件可内联，但"避免昂贵操作"的 guard 应保留

哪些该改，哪些不该改
可以简化
# 单层简单条件
def _build_cwd_info(cwd: str | None) -> str:
    if not cwd:
        return ""
    return f"<workspace-info>\n..."

# → 内联
def _build_cwd_info(cwd: str | None) -> str:
    return "" if not cwd \
        else f"<workspace-info>\n..."
不应该改
# 避免无意义 LLM 调用 — 语义清晰
if not new_memories or not candidates:
    return []
# 这个 guard 防止了昂贵的 LLM 去重操作
# 保留比压成表达式更好

# 避免文件 IO — 有副作用
if not workspace_root.exists():
    return None

不应该消除的 Guard Clause（~2,430 个）

以下类型的 guard 是代码质量的体现，消除它们会降低可读性或引入 bug

DB/缓存查询 — 真正可能是 None

record = store.get(id)
if record is None:  # DB 确实可能没有
    raise HTTPException(404, ...)
# 这个 None 不是类型问题，是业务现实

LLM / JSON 输出 — 不确定性边界

if isinstance(message, str):
    return message, []
# LLM 返回值类型不可控
# 这种 isinstance 是必要的运行时检查

可选服务 — 运行时可能未配置

if self._sandbox_provider is None:
    return LocalRuntime(...)
# 沙盒不一定启用，这不是 Optional 泛滥
# 而是真实的部署差异

避免昂贵操作 — 语义性 guard

if not new_memories or not candidates:
    return []  # 跳过无意义的 LLM 去重调用
# guard 的价值不在"消除分支"
# 而在"避免浪费"

推荐起手文件：software_factory.py

8+
重复 404 guard
20~30
单文件可消除
低
改动风险

同时做 Depends 抽取 + 请求体 Pydantic 化，效果最直观。改完可作为模板推广到 sessions.py、agent_board.py、skills.py。

投入产出比矩阵

优先级 手段 预估消除 难度 风险 收益 / 投入
P0 FastAPI Depends 合并 70~95 低~中 改 handler 签名
P0 Pydantic 声明式校验 45~60 低 400→422 状态码
P1 Optional 收窄 25~45 中~高 测试/旧会话兼容
P2 isinstance 重写 30~45 低 仅重写，低风险
P2 Early return 简化 20~30 低 可读性可能下降
合计 180~260 占 2,690 guard clause 的 7~10%

最终结论

可优化分支（第一轮）
  • 非 guard 的 3,056 个分支中，通过 dict dispatch、事件分发统一、Enum 替换等手段可消除 1,500~2,000 个
  • Top 8 文件集中治理可消除 185-290 个
Guard Clause（第二轮）
  • 2,690 个 guard clause 中仅 180-260 个适合消除
  • 最佳起手：software_factory.py 的 Depends + Pydantic 化
  • ~2,430 个是合法边界防御，不应消除
建议执行路径
software_factory.py Depends 示范 → routes 层 Pydantic 化 → 全局 EventDispatcher 基础设施 → Top 8 文件逐个治理 → Optional 收窄（按需）