边条件与处理器

边条件与处理器

学习目标

• 理解 EdgeConditionManager 的抽象与实现
• 掌握 Function 和 Keyword 两种条件评估方式
• 了解 Regex Extract 和 Function 两种数据变换处理器
• 能够配置复杂的边条件路由（如 ChatDev_v1 中的 Review → Fix/Approve 分支）

---

项目实践

条件评估架构

ChatDev 的边条件系统基于 EdgeConditionManager 抽象：

Function 条件

Function 条件调用预定义的 Python 函数：

condition:
  type: function
  config:
    name: code_fail    # 函数名

内置条件函数（位于 functions/edge/）：

函数名	行为
true	始终返回 True
always_false	始终返回 False
code_fail	检查是否包含代码错误标记
code_pass	检查代码是否通过
contains_keyword	检查是否包含特定关键词
need_reflection_loop	判断是否需要反思循环
should_stop_loop	判断是否应停止循环

Keyword 条件

Keyword 条件基于文本匹配：

condition:
  type: keyword
  config:
    any:                 # 包含任一关键词 → True
      - "ROUTE: Planner"
      - "FINISHED"
    none:                # 包含任一词 → False
      - "ERROR"
    regex:               # 正则匹配
      - "Result:.*PASS"
    case_sensitive: true

评估逻辑：

result = True
if any 列表非空:
    result = any(keyword in payload.text for keyword in config.any)
if none 列表非空:
    result = result and not any(keyword in payload.text for keyword in config.none)
if regex 列表非空:
    result = result and any(re.search(pattern, payload.text) for pattern in config.regex)

边处理器（Edge Processor）

边处理器在消息流经边时进行数据变换：

process:
  type: regex_extract
  config:
    pattern: "<result>(.*)</result>"
    group: 1
    mode: replace_content     # 替换内容 / metadata / data_block
    on_no_match: pass         # 不匹配时的行为
    default_value: null

Regex Extract Processor

三种模式：

模式	行为
replace_content	用匹配结果替换消息内容
metadata	将匹配结果写入消息元数据
data_block	将匹配结果写入消息数据块

不匹配时的处理：

on_no_match	行为
pass	原样传递消息
default	使用 default_value 替换
drop	丢弃消息（边不激活）

Function Processor

函数处理器调用 Python 函数对消息进行自定义变换：

process:
  type: function
  config:
    name: uppercase_payload    # 函数名

内置处理器（位于 functions/edge_processor/）：

函数名	行为
uppercase_payload	将消息内容转为大写

边的完整语义

一个边的所有属性组合：

- from: node_a
  to: node_b
  trigger: true                    # 是否评估此边
  condition: { ... }               # 条件评估
  carry_data: true                 # 是否携带输入数据
  keep_message: false              # 是否保留消息到目标节点上下文
  clear_context: false             # 是否清理目标节点的现有上下文
  clear_kept_context: false        # 是否清理之前 keep_message 保留的消息
  process: { ... }                 # 数据变换
  dynamic: { ... }                 # 动态分裂

属性的组合效果：

问题与规避

问题	表现	规避策略
条件永远为假	目标节点永不执行	确保至少一条入边 trigger: true 且 condition: 'true'
正则提取不匹配	on_no_match: drop 导致消息丢失	使用 pass 或 default 作为安全策略
关键词大小写错误	case_sensitive: true 导致匹配失败	检查实际输出的大小写
clear_context 误用	清理了必要的历史上下文	仅在需要隔离场景时使用
函数条件抛出异常	条件评估失败	函数中 try-except 捕获并返回 False

设计取舍

Function vs Keyword 条件

维度	Function	Keyword
表达能力	图灵完备	仅限文本匹配
编写难度	需编写 Python	零代码
可调试性	高（断点调试）	低（纯配置）
安全性	中（可执行任意代码）	高（仅匹配）
ChatDev 的选择	两者并存	两者并存

处理器失败的处理

ChatDev 选择 不阻断图执行 的策略：

try:
    return processor.transform(payload, ...)
except Exception as exc:
    log_manager.error(f"Edge payload processor failed: {exc}")
    return payload  # fallback 到原始消息

这保证了处理器 bug 不会导致整个工作流崩溃。

参考来源

• ChatDev Edge Conditions 源码
• ChatDev Edge Processors 源码