LangGraph PSE: Explicitly Modeling PSE Tri-role Collaboration with State Machines

🇨🇳 中文版

Why an Explicit State Machine?

Multi-agent frameworks follow two philosophies: implicit workflows (agents orchestrate freely) and explicit state machines (every state transition is observable, debuggable, and controllable).

CrewAI's sequential process is the former—agents execute in order, but you can't inspect intermediate states at runtime, can't inject custom logic at a node, and can't precisely control retry paths. When tasks get complex (e.g., a "generate → verify → correct" loop), black-box workflows become a debugging nightmare.

LangGraph's StateGraph offers an alternative: define the graph structure explicitly in code—nodes, edges, and conditional branches are all visible. langgraph-pse leverages this to turn the Planner → Specialist → Evaluator → Fix PSE pattern into an auditable architecture diagram:

This isn't an abstract description—it's the actual structure of build_graph() in graph.py. The flow is:

1. START → planner → specialist → evaluator
2. Evaluator decision:
   • Pass → END
   • Issues found → fix → back to evaluator (retry loop)

You can visualize this graph with LangGraph Studio and see state snapshots at every step.

Core Innovation: Merge Gate

The most unique design in langgraph-pse isn't the graph structure—it's the Evaluator node.

In traditional PSE frameworks, the Evaluator is just an LLM review role—asking the LLM to judge its own output. This has two fatal flaws:

1. LLM self-judgment is unreliable: LLMs tend to rate their own output highly ("self-preference bias")
2. Subjective issues cause infinite loops: The LLM can find new issues every round (wording, style, preferences), so Fix never converges

langgraph-pse solves this with a Merge Gate—two verification channels running in parallel, results merged:

def _make_evaluator_node(client, prompt: str, verify_fn: Optional[Callable]):
    """Merge Gate: LLM Review (first round only) + Programmatic verify_fn Hard Check (every round).

    - On the first round (attempts==0), call LLM to review output against real data;
      only run LLM on the first round to avoid infinite loops from subjective issues.
    - Programmatic verify_fn runs every round (deterministic, more reliable than
      LLM self-judgment, prevents hallucination).
    - Merge both into fictitious to drive fix retries.
    """
    def evaluator(state: PSEState) -> dict:
        # 1) LLM Review (first round only)
        eval_issues = []
        if attempts == 0 and prompt:
            # ... call LLM to review output against real data ...

        # 2) Programmatic Verification (every round)
        if verify_fn is not None:
            prog_bad, ok = verify_fn(state)

        all_bad = list(prog_bad) + list(eval_issues)  # merge

Key decisions:

	LLM Review	Programmatic verify_fn
Frequency	First round only	Every round
Capability	Macro-judgment (logical consistency, expression quality)	Micro hard-check (is the number correct, does the reference exist)
Risk	Subjective issues cause infinite loops	None (deterministic output)
Convergence guarantee	No	Yes—once Fix corrects all issues, verify_fn passes

Running LLM only on the first round is the critical convergence design: the first round gets LLM's macro review, subsequent rounds are driven solely by programmatic checks. This ensures the Fix loop must converge—once all hard-check issues are corrected, the process terminates.

verify_fn: The Framework's Soul Extension Point

build_graph() accepts a verify_fn parameter with signature (state) -> (bad: list, ok: list). This is what distinguishes the framework from crewai-pse (which focuses on article generation)—langgraph-pse was designed from the start as a universal quality-loop framework, with verify_fn as the soul of task adaptation.

The CRM-QA task demonstrates a real verify_fn implementation—verifying that numbers referenced in the report match the scan results:

def _verify_report(report: str, scan_json: dict) -> tuple[list, list]:
    """Numbers referenced in the report must be found in scan results.

    Design principle: prefer false negatives over false positives—as long as
    the report contains the true value within reasonable range, it passes.
    Avoid flagging LLM's natural language expressions (e.g., "about 6%",
    "accounts for 39%") as hallucinations that would cause fix death loops.
    """
    for f in findings:
        check = f["check"]
        target = f["count"]
        # Locate check in Markdown table row, extract number, compare
        if val == target:
            ok.append(f"{check} = {target}")
        else:
            bad.append(f"{check} report says {val} but actual is {target}")

The "prefer false negatives over false positives" principle is crucial—if the LLM writes "about 6%" and the real value is 5.8%, that shouldn't be flagged as an error. This tolerance for natural language expression is a design insight for programmatic verification in the LLM era.

State Transparency: PSEState Data Flow

The core challenge of PSE frameworks is observability of state as it flows between three roles. langgraph-pse uses a TypedDict to explicitly define every state field:

class PSEState(TypedDict, total=False):
    task_input: str          # Input: task context
    task_data: dict          # Input: additional data (e.g., scan results) for verify_fn
    plan: str                # Planner's execution plan
    artifact: str            # Specialist's output (report, corrected data, etc.)
    attempts: int            # Control flow: current retry count
    fictitious: list         # Control flow: issue list (programmatic + review merged)
    verified: list           # Control flow: passed items
    eval_issues: list        # Control flow: LLM review issues (first round only)
    max_retries: int         # Control flow: max retry count

Three layers at a glance:

• Input layer: task_input + task_data—data entering the framework
• Artifact layer: plan + artifact—Planner's plan, Specialist's output
• Control flow layer: attempts, fictitious, verified, eval_issues—driving retries and termination

Compared to implicit workflow frameworks, you don't need to dig through logs to guess "what did the LLM output?"—every field's value is queryable at any point in time.

Triple Sandbox: Defense in Depth for Toolchain Security

langgraph-pse's toolchain isn't a simple "give the agent a bash shell"—it's three layers of defense in depth:

Layer 1: Path Sandbox (read_file)

@tool("read_file")
def read_file(path: str) -> str:
    p = Path(path).resolve()
    if not str(p).startswith(str(_PROJECT_ROOT)):
        return f"[Error] Path exceeds project scope: {path}"

The agent can only read files within the project directory. Any ../../etc/passwd attempt gets intercepted.

Layer 2: Command Blacklist (run_bash)

_DANGEROUS_PATTERNS = [
    r"\brm\s+-rf\b", r"\bmkfs\b", r"\bdd\b\s+if=",
    r":\(\)\s*\{",                   # fork bomb
    r"curl\b[^\n]*\|\s*(sh|bash)",  # pipe to shell
    r"\bnc\b[^\n]*-e\b",            # reverse shell
    # ... 14 patterns total
]

Covers rm -rf, fork bombs, curl | sh, reverse shells, and other common attack patterns.

Layer 3: SQL Read-Only Sandbox (query_crm)

@tool("query_crm")
def query_crm(sql: str) -> str:
    if not sql.strip().lower().startswith("select"):
        return "[Rejected] Only SELECT queries allowed."
    if ";" in sql.strip().rstrip(";"):
        return "[Rejected] Only single SELECT, multi-statement not supported."
    con = sqlite3.connect(f"file:{db}?mode=ro&immutable=1", uri=True)

Three lines of defense: reject non-SELECT, reject multi-statement, SQLite immutable mode (even if the first two are bypassed, writes are impossible).

In Practice: CRM Data Quality Watchdog

langgraph-pse currently runs in the personal-crm project as a data quality watchdog. This scenario is completely different from "writing articles"—the artifact is a structured QA report, and verification checks numeric accuracy.

Usage:

# Deterministic scan (zero cost, no API key needed)
python tasks/crm-qa/run.py --db /path/to/crm.db

# LLM-generated natural language QA report + numeric verification
python tasks/crm-qa/run.py --db /path/to/crm.db --llm

The design philosophy of two modes:

• Default mode: Pure programmatic scan, outputs JSON findings, zero LLM cost. Suitable for CI/CD scheduled checks.
• –llm mode: On top of scan results, Specialist writes a natural language report, verify_fn checks that numbers referenced in the report match scan results. Suitable for human review.

This combination of "deterministic scan + LLM report + programmatic numeric verification" embodies langgraph-pse's design core: LLM handles expression, programs handle verification.

build_graph(): Composable Graph Construction

build_graph() is the framework's entry point, using parameter composition rather than inheritance to customize behavior:

graph = build_graph(
    task="crm-qa",              # loads tasks/crm-qa/prompts/*.md
    verify_fn=_verify_state,    # inject numeric verification
    use_planner=True,           # whether to include planning phase
    max_retries=3,              # max correction rounds
    provider="deepseek",        # LLM backend
)

Key design decisions:

• use_planner=False: Simple tasks can skip Planner and go straight to Specialist. Avoids unnecessary planning overhead for "look up a number" type tasks.
• verify_fn=None: Without programmatic verification, Evaluator degenerates to pure LLM review (first round only).
• tools=None: Defaults to read_file + run_bash; can be customized with a different tool list.

This "configuration via parameters" style is lighter than an inheritance hierarchy—you don't need to write a CRMQAGraph(StateGraph) subclass, just pass parameters.

Source Code Navigation

Module	File	Description
State Machine Core	graph.py	PSEState, node construction, merge gate, graph compilation
Triple Sandbox Tools	tools.py	read_file + run_bash + query_crm + crm_qa_scan
Configuration	config.py	Multi-provider environment variables
Model Creation	model.py	ChatOpenAI + exponential backoff retry
Prompt Loading	prompts.py	Dynamic loading of tasks//prompts/*.md
CRM-QA Entry	run.py	Scan + LLM report + verify_fn numeric check
QA Scanner	qa_scan.py	Deterministic data quality scan

Quick Start

# Install dependencies
uv sync

# Configure environment variables (see .env.example)
cp .env.example .env
# Edit .env, fill in OPENAI_API_KEY / OPENAI_BASE_URL / OPENAI_MODEL
# or AGNES_KEY / AGNES_BASE_URL / AGNES_MODEL

# Run deterministic scan (zero cost)
python tasks/crm-qa/run.py --db /path/to/crm.db

# Generate QA report with LLM + numeric verification
python tasks/crm-qa/run.py --db /path/to/crm.db --llm