一、引言
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
