Erishen

Building a PSE Tri-Role Closed Loop with AutoGen: A Retriable, Traceable Agent Collaboration Framework

🇨🇳 中文版

1. Introduction

autogen-pse is a Planner-Specialist-Evaluator (PSE) tri-role Agent collaboration framework built on Microsoft AutoGen. It implements state transitions via sequential speaking using RoundRobinGroupChat, combined with cycle-level loop control and token optimization strategies, forming an automated delivery pipeline that is retriable, traceable, and auditable.

2. Core Design

2.1 Design Decision 1: Using RoundRobinGroupChat Instead of Function Calling to Implement the PSE State Machine

Intuitive Approach: Define tool functions (e.g., plan(), execute(), evaluate()) for the Planner, Specialist, and Evaluator respectively, and use the LLM's Function Calling to select the next action, manually maintaining the state transition graph.

Actual Implementation: Directly use AutoGen's RoundRobinGroupChat (sequential speaking mode), coupled with custom termination_condition and max_turns to control the length of each round of dialogue. An outer while True cycle loop wraps this, dynamically modifying the current_task text injected into the conversation in run_task() based on the detection results from _detect_outcome().

# 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,
    )

Why This Is Better: It avoids the complexity of hand-writing a state machine—the LLM naturally understands "who should speak" (RoundRobin order), while the cycle-level control logic is concentrated in the orchestrator. By rewriting the task text, it implements divergent paths for "fix and retry" vs. "re-plan," resulting in cleaner and more extensible code.

2.2 Design Decision 2: Dual-Track Retry and Context Control—Pass Summaries for PARTIAL, Clear Context for FAIL

Intuitive Approach: Regardless of whether the outcome is PARTIAL or FAIL, pass the complete conversation history to the next cycle loop, allowing the LLM to see all context.

Actual Implementation: In _run_one_cycle(), the messages list is collected for trace recording, but what is passed to the next round is the rewritten current_task text (rather than the original conversation history). For PARTIAL, only result.summary (the judgment summary, truncated to 2000 characters) is passed; for FAIL, an instruction to "re-plan based on the original task" is passed, clearing the details of the previous discussion.

# 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)

And in run_task(), the handling logic for PARTIAL and FAIL is as follows:

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

Why This Is Better: It prevents token explosion—as the number of cycles increases, the token count of the complete conversation history grows linearly, wasting costs and potentially exceeding the model's context window. By passing only structured summaries or rewritten tasks, each round maintains constant token overhead while ensuring precise delivery of the Evaluator's feedback (PARTIAL retains problem details for fixing, FAIL requires a complete restart).

2.3 Design Decision 3: An Extensible Task Architecture—Registration, Rule Injection, and Post-Processing Correction

Intuitive Approach: Hardcode a set of prompts for each task, or switch configurations for different tasks via command-line arguments.

Actual Implementation: Three-layer design—

  1. Task Registry _registry.json centrally manages all tasks, with CLI dispatching via subcommands in cli.py.
  2. Prompt Loader prompts.py supports dynamic loading of tasks/<name>/prompts/*.md based on the task parameter. If not found, it falls back to the generic demo prompt, and replaces placeholders like {INVESTMENT_RULES} / {STOP_LOSS_RULES} with task-specific rule file contents via _inject_rules().
  3. Post-Processing Pipeline In portfolio-review/run.py, extract_review_from_trace() reverse-extracts the Specialist's final conclusion from the trace JSON and saves it as a Markdown file. sanitize_review() uses a rule engine to forcibly correct LLM violations (e.g., accidental selling of defensive core positions, erratic advice for small holdings).

Prompt Loading and Rule Injection (src/autogen_pse/prompts.py):

def _load_private_rules(task_dir: Path, name: str) -> str:
    """Load task-private rule files, returning an empty string if they don't exist."""
    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:
    """Replace {INVESTMENT_RULES} / {STOP_LOSS_RULES} placeholders with private rule file contents."""
    rules = _load_private_rules(task_dir, name)
    if not rules:
        return template

    # Replace possible placeholders
    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:
    """Load the system prompt for a specified role."""
    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")

    # Fall back to generic prompts
    prompt_path = PROMPTS_DIR / "demo" / "prompts" / f"{name}.md"
    if not prompt_path.exists():
        raise FileNotFoundError(f"Prompt file does not exist: {prompt_path}")
    return prompt_path.read_text(encoding="utf-8")

Post-Processing Pipeline (tasks/portfolio-review/run.py):

extract_review_from_trace() reverse-extracts the Specialist's final conclusion from the trace JSON and saves it as a Markdown file; sanitize_review() uses a rule engine to forcibly correct LLM violations, for example:

  • Sell recommendations for certain core positions → Change to hold/observe (unless the stop-loss threshold is triggered)
  • Action recommendations for small positions → Remove (too much noise, no practical significance)

The core philosophy of post-processing is: never trust the LLM's final output—use deterministic rules as a safety net to ensure deliverables meet predefined constraints.

Why This Is Better: Adding a new task requires only three steps (create directory, write three prompts, write run.py), without modifying the core engine. Private rule files (such as stop-loss rules) do not enter the prompt template, keeping the template generic while allowing task-level customization. Post-processing correction addresses the final link of LLM unreliability, forming a complete closed loop of "generic engine + private tasks + regulated output."

3. Web Dashboard

Beyond CLI and Makefile entry points, autogen-pse also provides a FastAPI + SSE-based Web Dashboard for real-time observation of PSE execution and history.

Key features:

  • One-Click Execution: Trigger tasks via the Web interface, with SSE streaming output making each Agent's thinking step visible in real time
  • Execution History: Automatically records traces from each run, with filtering by time and verdict
  • Trace Details: Expand the full conversation for each cycle, with per-Agent Token consumption breakdown
  • Auto Detect Prepare: The Web layer automatically detects whether a task requires data preparation—no manual prepare needed

Technically, web_server.py resides in the project root, using FastAPI to provide REST API + SSE endpoints, with a Vite + React frontend. Start with:

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

4. Source Code Navigation

Module File Description
Core Orchestration orchestrator.py Full lifecycle management of create_pse_team() + run_task(), cycle control, token tracking, trace writing
Agent Factory agents.py Creation of AssistantAgent for the three roles, with differences in tool binding (Specialist has only read_file, Evaluator has run_pytest/ruff)
Prompt System prompts.py Task-level prompt loading + private rule injection (replacement of placeholders like {STOP_LOSS_RULES})
Utility Functions tools.py TokenTracker/TokenReport, bash/read_file/run_pytest/run_ruff, RAG knowledge base retrieval
CLI Entry cli.py Subcommand dispatch, driven by task registry
Web Service web_server.py FastAPI + SSE streaming output, automatic detection of prepare timing, trace query API
Demo Task tasks/demo/run.py + tasks/demo/prompts/ Quick sort example, minimal runnable task
Investment Task tasks/portfolio-review/run.py + prepare.py Full production-grade task, including knowledge base injection, trace extraction, sanitize post-processing

5. Quick Start

git clone https://github.com/erishen/autogen-pse.git
cd autogen-pse
cp .env.example .env          # Configure API Key and external paths
uv sync                        # Install dependencies
make demo                      # Run quicksort demo

To run the investment weekly report task:

make review                    # Run PSE investment analysis
make serve                     # Start Web Dashboard (http://localhost:8080)

CLI Interaction Example:

python cli.py list             # List all available tasks
python cli.py run portfolio-review   # Run investment analysis task
python cli.py trace -n 5       # View trace of the last 5 executions
首页 简历 关于 隐私政策

© 2026 Erishen
沪ICP备2024079226号-1   沪公网安备31010502007082号