基于 AutoGen 构建 PSE 三角色闭环：一个可重试、可追溯的 Agent 协作框架

🇬🇧 English Version

一、引言

autogen-pse 是一个基于 Microsoft AutoGen 构建的 Planner-Specialist-Evaluator（PSE） 三角色 Agent 协作框架。它通过顺序发言的 RoundRobinGroupChat 实现状态流转，配合 cycle 级别的循环控制与 token 优化策略，形成一个可重试、可追溯、可审计的自动化交付管线。

二、核心设计

2.1 设计决策 1：用 `RoundRobinGroupChat` 替代 Function Calling 实现 PSE 状态机

直觉会怎么做：为 Planner、Specialist、Evaluator 分别定义工具函数（如 plan(), execute(), evaluate()），通过 LLM 的 Function Calling 来选择下一步动作，手动维护状态转移图。

实际怎么做的：直接使用 AutoGen 的 RoundRobinGroupChat（顺序发言模式），配合自定义的 termination_condition 和 max_turns 控制每轮对话长度。外层再套一个 while True 的 cycle 循环，在 run_task() 中根据 _detect_outcome() 的检测结果动态修改下一轮的 current_task 文本注入到对话中。

# src/autogen_pse/orchestrator.py
def create_pse_team(
    model_client: Optional[OpenAIChatCompletionClient] = None,
    task: Optional[str] = None,
) -> RoundRobinGroupChat:
    if model_client is None:
        model_client = _create_model_client()

    planner = create_planner(model_client, task)
    specialist = create_specialist(model_client, task)
    evaluator = create_evaluator(model_client, task)

    text_term = TextMentionTermination("交付完成") | TextMentionTermination("BLOCKED")
    return RoundRobinGroupChat(
        participants=[planner, specialist, evaluator],
        termination_condition=text_term | ExternalTermination(),
        max_turns=TURNS_PER_CYCLE,
    )

为什么更好：避免了手写状态机的复杂性——LLM 天然理解"谁该说话"（RoundRobin 顺序），而 cycle 级别的控制逻辑集中在 orchestrator 中，通过改写 task 文本实现"修复重试" vs "重新规划"的分歧路径，代码更简洁且易于扩展。

2.2 设计决策 2：双轨重试与上下文控制——PARTIAL 传摘要、FAIL 清上下文

直觉会怎么做：无论 PARTIAL 还是 FAIL，都将完整的对话历史传递给下一轮循环，让 LLM 看到全部上下文。

实际怎么做的：在 _run_one_cycle() 中，messages 列表被收集用于 trace 记录，但传给下一轮的是改写后的 current_task 文本（而不是原始对话历史）。对于 PARTIAL，只传递 result.summary（判决摘要，截断至 2000 字）；对于 FAIL，传递"基于原始任务重新制定计划"的指令，清空上轮讨论细节。

# src/autogen_pse/orchestrator.py
async def _run_one_cycle(
    team: RoundRobinGroupChat,
    task: str,
    verbose: bool,
) -> CycleResult:
    tracker = TokenTracker()
    messages = []

    async for msg in team.run_stream(task=task):
        tracker.feed(msg)
        if isinstance(msg, TaskResult):
            continue
        messages.append(msg)
        if verbose:
            _print_message(msg)

    outcome, reason, summary = _detect_outcome(messages)
    chat_log = []
    for m in messages:
        if isinstance(m, TextMessage):
            chat_log.append({"source": m.source, "content": m.content})
    return CycleResult(outcome, summary, tracker.report, chat_log, reason=reason)

而在 run_task() 中，PARTIAL 和 FAIL 的处理逻辑如下：

if result.outcome == "PARTIAL":
    partial_count += 1
    if partial_count > MAX_PARTIAL_RETRIES:
        current_task = (
            f"连续 PARTIAL {partial_count} 次，已达上限。"
            f"最近原因: {result.reason}。"
            f"请评估是否仍有可交付内容，或宣布 BLOCKED。\n\n"
            f"原始任务: {task}\n\n"
            f"最近判决: {result.summary}"
        )
        continue
    current_task = (
        f"上一轮 Evaluator 判决 PARTIAL (原因: {result.reason})。"
        f"请修复以下问题后重新提交：\n\n{result.summary}"
    )
    continue

if result.outcome == "FAIL":
    fail_count += 1
    fail_rem = MAX_FAIL_RETRIES - fail_count
    if fail_rem < 0:
        if verbose:
            print(f"\n❌ 连续 FAIL {MAX_FAIL_RETRIES}+ 次，强制 BLOCKED")
        break
    current_task = (
        f"上一轮被判 FAIL (原因: {result.reason})。请基于原始任务重新制定计划。"
        f"（剩余重试次数: {fail_rem}）\n\n"
        f"原始任务: {task}\n\n"
        f"失败原因: {result.summary}"
    )
    continue

为什么更好：防止 token 爆炸——随着循环次数增加，完整对话历史的 token 量会线性增长，既浪费成本又可能超出模型上下文窗口。通过只传递结构化摘要/改写任务，每轮保持恒定 token 开销，同时确保 Evaluator 的反馈精准传递（PARTIAL 保留问题细节以便修复，FAIL 要求彻底重来）。

2.3 设计决策 3：可扩展的任务架构——注册、规则注入与后处理修正

直觉会怎么做：每个任务硬编码一套 prompt，或通过命令行参数切换不同任务的配置。

实际怎么做的：三层设计——

任务注册表 _registry.json 统一管理所有任务，CLI 通过 cli.py 的子命令调度
prompt 加载器 prompts.py 支持按 task 参数动态加载 tasks/<name>/prompts/*.md，不存在时回退到 demo 通用 prompt，并通过 _inject_rules() 将 {INVESTMENT_RULES} / {STOP_LOSS_RULES} 占位符替换为任务私密的规则文件内容
后处理管道 在 portfolio-review/run.py 中，extract_review_from_trace() 从 trace JSON 中逆向提取 Specialist 的最终结论并保存为 Markdown 文件，sanitize_review() 用规则引擎强制修正 LLM 的违规输出（如防御底仓误卖、小额持仓乱建议）

Prompt 加载与规则注入（src/autogen_pse/prompts.py）：

def _load_private_rules(task_dir: Path, name: str) -> str:
    """加载任务私密规则文件，不存在时返回空字符串。"""
    rules_path = task_dir / f"{name}_rules.md"
    if rules_path.exists():
        return rules_path.read_text(encoding="utf-8").strip()
    return ""


def _inject_rules(template: str, name: str, task_dir: Path) -> str:
    """将 {INVESTMENT_RULES} / {STOP_LOSS_RULES} 占位符替换为私密规则文件内容。"""
    rules = _load_private_rules(task_dir, name)
    if not rules:
        return template

    # 替换可能的占位符
    for placeholder in ("INVESTMENT_RULES", "STOP_LOSS_RULES"):
        key = "{" + placeholder + "}"
        template = template.replace(key, rules)
    return template


def load_prompt(name: str, task: str | None = None) -> str:
    """加载指定角色的系统提示词。"""
    if task:
        task_dir = PROMPTS_DIR / task
        prompt_path = task_dir / "prompts" / f"{name}.md"
        if prompt_path.exists():
            template = prompt_path.read_text(encoding="utf-8")
            return _inject_rules(template, name, task_dir / "prompts")

    # 回退到通用提示词
    prompt_path = PROMPTS_DIR / "demo" / "prompts" / f"{name}.md"
    if not prompt_path.exists():
        raise FileNotFoundError(f"提示词文件不存在: {prompt_path}")
    return prompt_path.read_text(encoding="utf-8")

后处理管道（tasks/portfolio-review/run.py）：

extract_review_from_trace() 从 trace JSON 中逆向提取 Specialist 的最终结论并保存为 Markdown 文件；sanitize_review() 用规则引擎强制修正 LLM 的违规输出，例如：

某类核心持仓的卖出建议 → 改为持有观察（除非触发止损线）
小额持仓的操作建议 → 移除（噪音过多，无实际意义）

后处理的核心思路是：不信任 LLM 的最终输出，用确定性规则兜底修正，确保交付物符合预设约束。

为什么更好：新增任务只需三步（建目录、写三个 prompt、写 run.py），无需改动核心引擎。私密规则文件（如止损规则）不进入 prompt 模板，保持模板通用性同时允许任务级定制。后处理修正弥补了 LLM 不可靠性的最后一环，形成"引擎通用化 + 任务私有化 + 输出规则化"的完整闭环。

三、Web Dashboard

除了 CLI 和 Makefile 入口，autogen-pse 还提供了一个基于 FastAPI + SSE 的 Web Dashboard，支持实时查看 PSE 执行过程和历史记录。

核心功能：

一键执行：通过 Web 界面触发任务，SSE 流式输出让 Agent 的每一步思考过程实时可见
执行历史：自动记录每次运行的 trace，支持按时间、判决结果筛选查看
Trace 详情：展开每一轮的完整对话，按 Agent 维度统计 Token 消耗
自动 detect prepare：Web 层自动检测任务是否需要数据准备，无需手动运行 prepare

技术实现上，web_server.py 位于项目根目录，使用 FastAPI 提供 REST API + SSE 端点，前端通过 Vite + React 构建。启动方式：

make serve    # 启动 Web Dashboard (http://localhost:8080)

四、源码导航

模块	文件	说明
核心编排	orchestrator.py	`create_pse_team()` + `run_task()` 全生命周期管理，cycle 控制、token 追踪、trace 写入
Agent 工厂	agents.py	三个角色的 `AssistantAgent` 创建，工具绑定差异（Specialist 只有 read_file，Evaluator 有 run_pytest/ruff）
Prompt 系统	prompts.py	任务级 prompt 加载 + 私密规则注入（`{STOP_LOSS_RULES}` 等占位符替换）
工具函数	tools.py	TokenTracker/TokenReport、bash/read_file/run_pytest/run_ruff、RAG 知识库检索
CLI 入口	cli.py	子命令调度，任务注册表驱动
Web 服务	web_server.py	FastAPI + SSE 流式输出，自动 detect prepare 时机，trace 查询 API
Demo 任务	tasks/demo/run.py + tasks/demo/prompts/	快速排序示例，最小可运行任务
投资任务	tasks/portfolio-review/run.py + prepare.py	完整生产级任务，含知识库注入、trace 提取、sanitize 后处理

五、快速开始

git clone https://github.com/erishen/autogen-pse.git
cd autogen-pse
cp .env.example .env          # 配置 API Key 和外部路径
uv sync                        # 安装依赖
make demo                      # 运行 quicksort demo

如需运行投资周报任务：

make review                    # 运行 PSE 投资分析
make serve                     # 启动 Web Dashboard (http://localhost:8080)

CLI 交互示例：

python cli.py list             # 列出所有可用任务
python cli.py run portfolio-review   # 运行投资分析任务
python cli.py trace -n 5       # 查看最近 5 次执行 trace

基于 AutoGen 构建 PSE 三角色闭环：一个可重试、可追溯的 Agent 协作框架

一、引言

二、核心设计

2.1 设计决策 1：用 RoundRobinGroupChat 替代 Function Calling 实现 PSE 状态机

2.2 设计决策 2：双轨重试与上下文控制——PARTIAL 传摘要、FAIL 清上下文

2.3 设计决策 3：可扩展的任务架构——注册、规则注入与后处理修正

三、Web Dashboard

四、源码导航

五、快速开始

评论

发表回复 取消回复

更多文章

基于 AutoGen 构建 PSE 三角色闭环：一个可重试、可追溯的 Agent 协作框架

AI-Analyze：五个不显而易见的设计决策

给 AI 助手搭一个工具服务器：Lobster 的架构思路

我如何从零构建一个生产级 React SSR 框架

2.1 设计决策 1：用 `RoundRobinGroupChat` 替代 Function Calling 实现 PSE 状态机

发表回复取消回复