#4962 AI Agent Team

#4962 AI Agent Team

#4962 I need an AI Agent Team with manager (orchestrator), finder, analyzer, recommender, implementer, next steps, as well as one or more critics

To build an effective multi-agent system, you need clear role boundaries, structured handoffs (ideally in JSON or strict Markdown schemas), and a feedback loop so the Critic can reject subpar work before it reaches the final stage.

Below is a production-ready architecture and system prompt blueprint for your 7-agent team. You can implement this in frameworks like LangGraph, CrewAI, AutoGen, or a custom Python loop.

1. System Architecture & Workflow

Instead of a purely linear pipeline, use a Hub-and-Spoke model controlled by the Manager (Orchestrator) with an embedded Critic Loop:

[User Goal] 
     │
     ▼
┌──────────────────────────────────────────────────────────┐
│                 1. MANAGER (Orchestrator)                │
│  Decomposes goal, delegates tasks, manages state/retries │
└────┬──────────┬──────────────┬───────────────┬───────────┘
     │          │              │               │
     ▼          ▼              ▼               ▼
[2. Finder] [3. Analyzer] [4. Recommender] [5. Implementer]
     │          │              │               │
     └──────────┴──────────────┴───────────────┘
                        │
                        ▼ (Every major output passes through)
               [7. CRITIC(S) (QA/Red Team)]
                        │
          ┌─────────────┴─────────────┐
          ▼ (Reject + Feedback)       ▼ (Approve)
   [Back to Agent]            [6. Next Steps] ──► [Final Output]

2. Agent Roles & System Prompts

Use these system prompts as the core instructions for each agent.

1. Manager (Orchestrator)

• Role: Project lead, state tracker, and router. Never does the grunt work; only plans, delegates, and synthesizes.

• System Prompt:

  You are the Orchestrator of a specialized AI agent team. Your job is to break down the user's objective into discrete tasks and route them to the appropriate specialist: Finder, Analyzer, Recommender, Implementer, Critic, and Next Steps.

  Rules:

  1. Never generate domain research, analysis, or code yourself.

  2. Enforce quality gates: route outputs from the Analyzer, Recommender, and Implementer to the Critic before moving to the next phase.

  3. If the Critic rejects an output, send it back to the originating agent with the Critic's exact feedback. Cap revision loops at 2 iterations to prevent infinite loops.

  4. Output your routing decisions in structured JSON: {"next_agent": "...", "task_instructions": "...", "context_needed": "..."}.

2. Finder (Researcher / Data Gatherer)

• Role: Information retrieval, fact-finding, and raw data extraction.

• Tools Needed: Web search, RAG/Vector DB, SQL/API connectors.

• System Prompt:

  You are the Finder. Your sole purpose is to gather accurate, verifiable, and raw facts, data, documentation, or context required by the Manager.

  Rules:

  1. Do not interpret, editorialize, or make recommendations. Stick strictly to retrieved facts.

  2. Always cite your sources, URLs, file names, or database tables.

  3. Explicitly state what information you could not find so downstream agents do not hallucinate missing data.

3. Analyzer (Synthesizer / Diagnostician)

• Role: Pattern recognition, root-cause analysis, SWOT, and data structuring.

• Tools Needed: Code interpreter (Python/Pandas) for math/data crunching.

• System Prompt:

  You are the Analyzer. You take raw data provided by the Finder and turn it into structured insights. Identify patterns, anomalies, constraints, correlations, and root causes.

  Rules:

  1. Base every claim strictly on the Finder's data. If data is insufficient, flag it immediately.

  2. Separate confirmed facts from your analytical inferences.

  3. Do not propose solutions or action plans—your job is strictly to diagnose the current state and explain why things are happening.

4. Recommender (Strategist)

• Role: Option generation, trade-off evaluation, and strategic decision-making.

• System Prompt:

  You are the Recommender. Based on the Analyzer's findings, generate viable strategic options and recommend the optimal path forward.

  Rules:

  1. Present 2–3 distinct options, detailing the pros, cons, risks, and resource costs of each.

  2. Clearly declare one primary recommendation and justify why it outperforms the alternatives given the user's constraints.

  3. Ensure your recommendation is actionable by the Implementer.

5. Implementer (Builder / Executor)

• Role: Drafting copy, writing code, building configurations, or creating the final deliverable.

• Tools Needed: Code execution sandbox, file writer, API integrations.

• System Prompt:

  You are the Implementer. Your job is to execute the Recommender's chosen strategy and produce the tangible deliverable (e.g., code, copy, financial model, workflow configuration).

  Rules:

  1. Follow the approved recommendation strictly; do not change the strategy mid-execution.

  2. Produce complete, production-ready artifacts—no placeholders (e.g., avoid // TODO: implement this).

  3. Document any assumptions made during execution.

6. Next Steps (Operations / Handoff Lead)

• Role: Packaging the final output, defining immediate human actions, and setting success metrics.

• System Prompt:

  You are the Next Steps Coordinator. Once the Critic approves the Implementer's work, you package the final response for the user.

  Rules:

  1. Summarize the final deliverable clearly.

  2. Provide a prioritized, chronological checklist of immediate actions the user needs to take (e.g., deployment steps, approvals, follow-ups).

  3. Define 2–3 concrete KPIs or verification checks so the user knows the implementation succeeded.

3. Designing the Critic(s)

Using two specialized Critics rather than one generic critic dramatically reduces hallucinations and blind spots:

Critic Type	Focus Area	When to Trigger
Logic & Fact Critic	Hallucinations, unsupported leaps in logic, missing edge cases, data discrepancies.	After Finder and Analyzer.
Execution & Risk Critic	Security flaws, scalability bottlenecks, tone/brand alignment, constraint violations.	After Recommender and Implementer.

Critic System Prompt Template:

You are the [Logic & Fact / Execution & Risk] Critic. Your job is to ruthlessly evaluate the previous agent's output against the user's original goal and established facts.

Rules:

1. Never rewrite or fix the work yourself; only critique it.

2. Check for: [Insert focus areas: e.g., logical fallacies, unverified claims, security risks, or incomplete code].

3. You must output a strict JSON verdict:

   {"status": "APPROVED" | "REJECTED", "severity": "LOW" | "HIGH", "critique": ["Specific issue 1", "Specific issue 2"], "required_fixes": "Exact instructions on what must change"}

4. Only mark "APPROVED" if there are zero high-severity flaws.

4. Shared State Schema (How They Communicate)

To keep the team from losing context or blowing up your token budget, pass a single Shared State Object between agents rather than dumping the entire chat history into every prompt:

{
  "user_goal": "Original prompt from user",
  "current_phase": "ANALYSIS",
  "iteration_count": 1,
  "artifacts": {
    "finder_data": "...",
    "analyzer_report": "...",
    "recommendations": "...",
    "implementation": "...",
    "next_steps": "..."
  },
  "critic_feedback": {
    "status": "REJECTED",
    "notes": "Option 2 exceeds the stated budget constraint."
  }
}

Recommended Model Allocation (Cost vs. Intelligence)

• Manager, Analyzer, Recommender, Critics: High-reasoning models (e.g., Gemini 2.5 Pro, Claude 3.7 Sonnet, or o3-mini) to prevent routing errors and catch subtle flaws.

• Finder & Next Steps: Fast, cost-effective models (e.g., Gemini 2.5 Flash or GPT-4o-mini) since their tasks are structured extraction and formatting.

• Implementer: Best-in-class domain model (e.g., Claude 3.7 Sonnet for coding, Gemini 2.5 Pro for long-context synthesis).

Python using Ollama

To run this 7-agent system locally in Python with Ollama, the cleanest approach is using the official ollama Python SDK paired with Pydantic. Ollama supports native Structured Outputs via the format parameter, which constrains the local LLM to output valid JSON matching your exact Pydantic schema—preventing formatting crashes during agent handoffs.

1. Setup & Local Models

Install the dependencies:

Bash

pip install ollama pydantic

Pull the recommended local open-weight models. Splitting the workload across a reasoning model and a fast model keeps VRAM usage manageable and speeds up execution:

Bash

# High-reasoning model for Manager, Analyzer, Recommender, Implementer & Critics
ollama pull qwen3:14b   # (or gpt-oss:20b / qwen3:30b if you have 24GB+ VRAM)

# Fast model for Finder and Next Steps formatting
ollama pull qwen3:8b

2. Complete Python Implementation (agent_team.py)

This script implements the Hub-and-Spoke Orchestrator, the Shared State, all 6 specialist agents, and the 2 specialized Critics with automatic retry loops.

import json
from typing import Literal, List, Optional
from pydantic import BaseModel, Field
from ollama import chat

# ==========================================
# 1. CONFIGURATION & MODELS
# ==========================================
REASONING_MODEL = "qwen3:14b"  # Manager, Analyzer, Recommender, Implementer, Critics
FAST_MODEL = "qwen3:8b"        # Finder, Next Steps
MAX_RETRIES = 2                # Prevent infinite Critic rejection loops

# ==========================================
# 2. PYDANTIC SCHEMAS (Structured Handoffs)
# ==========================================
class ManagerDecision(BaseModel):
    next_agent: Literal[
        "finder", "analyzer", "recommender", "implementer", "next_steps", "DONE"
    ]
    task_instructions: str = Field(description="Specific instructions for the chosen agent.")
    reasoning: str = Field(description="Why this agent is next.")

class CriticVerdict(BaseModel):
    status: Literal["APPROVED", "REJECTED"]
    severity: Literal["LOW", "HIGH"]
    critique: List[str] = Field(description="Specific flaws, hallucinations, or risks found.")
    required_fixes: str = Field(description="Exact instructions on what the agent must fix.")

class SharedState(BaseModel):
    user_goal: str
    finder_data: str = ""
    analyzer_report: str = ""
    recommendations: str = ""
    implementation: str = ""
    next_steps: str = ""
    history_log: List[str] = []

# ==========================================
# 3. CORE OLLAMA HELPER
# ==========================================
def run_llm(model: str, system_prompt: str, user_prompt: str, schema=None):
    """Calls local Ollama with optional Pydantic schema enforcement."""
    kwargs = {
        "model": model,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        # Temperature 0 ensures deterministic routing & JSON compliance
        "options": {"temperature": 0.0},
    }
    if schema:
        kwargs["format"] = schema.model_json_schema()

    response = chat(**kwargs)
    content = response.message.content

    if schema:
        return schema.model_validate_json(content)
    return content

# ==========================================
# 4. AGENT DEFINITIONS
# ==========================================
def manager_agent(state: SharedState) -> ManagerDecision:
    system_prompt = """You are the Orchestrator of an AI agent team.
    Inspect the current SharedState and decide which specialist runs next:
    1. 'finder' (if raw facts/context are missing)
    2. 'analyzer' (if finder_data exists, but analyzer_report is empty)
    3. 'recommender' (if analyzer_report exists, but recommendations are empty)
    4. 'implementer' (if recommendations exist, but implementation is empty)
    5. 'next_steps' (if implementation is complete, to package final handoff)
    6. 'DONE' (ONLY after next_steps is populated)
    Never do the work yourself. Output valid JSON matching the schema."""

    prompt = f"""
    USER GOAL: {state.user_goal}
    CURRENT ARTIFACTS:
    - Finder Data: {'POPULATED' if state.finder_data else 'EMPTY'}
    - Analyzer Report: {'POPULATED' if state.analyzer_report else 'EMPTY'}
    - Recommendations: {'POPULATED' if state.recommendations else 'EMPTY'}
    - Implementation: {'POPULATED' if state.implementation else 'EMPTY'}
    - Next Steps: {'POPULATED' if state.next_steps else 'EMPTY'}
    """
    return run_llm(REASONING_MODEL, system_prompt, prompt, schema=ManagerDecision)

def finder_agent(state: SharedState, instructions: str) -> str:
    system_prompt = """You are the Finder. Gather accurate facts, constraints, and context.
    Do NOT interpret or recommend solutions. Explicitly note any missing information."""
    prompt = f"Goal: {state.user_goal}\nInstructions: {instructions}"
    return run_llm(FAST_MODEL, system_prompt, prompt)

def analyzer_agent(state: SharedState, instructions: str, feedback: str = "") -> str:
    system_prompt = """You are the Analyzer. Synthesize Finder data into patterns, root causes, and constraints.
    Separate confirmed facts from inferences. Do NOT propose solutions."""
    prompt = f"Goal: {state.user_goal}\nInstructions: {instructions}\nFinder Data:\n{state.finder_data}"
    if feedback:
        prompt += f"\n\n[CRITICAL REVISION REQUIRED - FIX THESE ISSUES]:\n{feedback}"
    return run_llm(REASONING_MODEL, system_prompt, prompt)

def recommender_agent(state: SharedState, instructions: str, feedback: str = "") -> str:
    system_prompt = """You are the Recommender. Propose 2-3 strategic options with pros, cons, and costs.
    Declare ONE primary recommendation and justify why it is optimal for the Implementer."""
    prompt = f"Goal: {state.user_goal}\nInstructions: {instructions}\nAnalysis:\n{state.analyzer_report}"
    if feedback:
        prompt += f"\n\n[CRITICAL REVISION REQUIRED - FIX THESE ISSUES]:\n{feedback}"
    return run_llm(REASONING_MODEL, system_prompt, prompt)

def implementer_agent(state: SharedState, instructions: str, feedback: str = "") -> str:
    system_prompt = """You are the Implementer. Execute the Recommender's chosen strategy completely.
    Produce production-ready code, copy, or plans with NO placeholders or TODOs."""
    prompt = f"Goal: {state.user_goal}\nInstructions: {instructions}\nApproved Strategy:\n{state.recommendations}"
    if feedback:
        prompt += f"\n\n[CRITICAL REVISION REQUIRED - FIX THESE ISSUES]:\n{feedback}"
    return run_llm(REASONING_MODEL, system_prompt, prompt)

def next_steps_agent(state: SharedState, instructions: str) -> str:
    system_prompt = """You are the Next Steps Coordinator. Package the final deliverable for the user.
    Include: (1) Executive Summary, (2) Chronological User Action Checklist, (3) 2-3 Verification KPIs."""
    prompt = f"Goal: {state.user_goal}\nImplementation:\n{state.implementation}"
    return run_llm(FAST_MODEL, system_prompt, prompt)

def critic_agent(critic_type: Literal["Logic & Fact", "Execution & Risk"], state: SharedState, artifact_to_review: str) -> CriticVerdict:
    focus = (
        "hallucinations, unsupported leaps in logic, and data contradictions"
        if critic_type == "Logic & Fact"
        else "security flaws, edge cases, missing details, and constraint violations"
    )
    system_prompt = f"""You are the {critic_type} Critic. Ruthlessly evaluate the output for {focus}.
    Only mark APPROVED if there are zero HIGH severity flaws. Output strict JSON."""
    prompt = f"Original Goal: {state.user_goal}\n\nOutput to Review:\n{artifact_to_review}"
    return run_llm(REASONING_MODEL, system_prompt, prompt, schema=CriticVerdict)

# ==========================================
# 5. CRITIC FEEDBACK LOOP WRAPPER
# ==========================================
def run_with_critic(agent_fn, critic_type: str, state: SharedState, instructions: str) -> str:
    feedback = ""
    output = ""
    for attempt in range(MAX_RETRIES + 1):
        output = agent_fn(state, instructions, feedback)
        print(f"  🔍 [{critic_type} Critic] Reviewing output (Attempt {attempt + 1})...")
        verdict = critic_agent(critic_type, state, output)

        if verdict.status == "APPROVED":
            print(f"  ✅ [{critic_type} Critic] APPROVED.")
            return output

        print(f"  ❌ [{critic_type} Critic] REJECTED ({verdict.severity}): {verdict.critique}")
        feedback = f"Critique: {verdict.critique}\nRequired Fixes: {verdict.required_fixes}"

    print(f"  ⚠️ Max retries reached. Proceeding with best available version.")
    return output

# ==========================================
# 6. MAIN ORCHESTRATION LOOP
# ==========================================
def run_team(user_goal: str) -> SharedState:
    state = SharedState(user_goal=user_goal)
    print(f"🚀 Starting AI Agent Team for goal: '{user_goal}'\n")

    while True:
        decision = manager_agent(state)
        print(f"🧠 [Manager] -> Routing to: {decision.next_agent.upper()} ({decision.reasoning})")

        if decision.next_agent == "DONE":
            break
        elif decision.next_agent == "finder":
            state.finder_data = finder_agent(state, decision.task_instructions)
        elif decision.next_agent == "analyzer":
            state.analyzer_report = run_with_critic(
                analyzer_agent, "Logic & Fact", state, decision.task_instructions
            )
        elif decision.next_agent == "recommender":
            state.recommendations = run_with_critic(
                recommender_agent, "Execution & Risk", state, decision.task_instructions
            )
        elif decision.next_agent == "implementer":
            state.implementation = run_with_critic(
                implementer_agent, "Execution & Risk", state, decision.task_instructions
            )
        elif decision.next_agent == "next_steps":
            state.next_steps = next_steps_agent(state, decision.task_instructions)

    return state

if __name__ == "__main__":
    goal = "Design a rate-limiting middleware in Python for a FastAPI app handling 5,000 requests/minute."
    final_state = run_team(goal)

    print("\n" + "="*60)
    print("FINAL IMPLEMENTATION:\n", final_state.implementation)
    print("\n" + "="*60)
    print("NEXT STEPS & KPIs:\n", final_state.next_steps)

Key Local-LLM Optimizations Included

• Native Pydantic Schema Enforcement (format=Schema.model_json_schema()): Local models occasionally break JSON formatting when asked to output raw JSON via prompt alone. Passing the Pydantic schema directly to Ollama guarantees ManagerDecision and CriticVerdict never crash your routing loop.

• VRAM Model Swapping Prevention: Keep the number of distinct Ollama model tags to 1 or 2 (qwen3:14b and qwen3:8b). If you assign 7 different models to 7 agents, Ollama will spend most of its runtime unloading and reloading model weights from disk into VRAM.

• Temperature 0.0: Eliminates creative drift when the Manager is routing or when Critics are evaluating binary pass/fail criteria.

Show me how to add real tools (like DuckDuckGo web search for Finder and local file writing for Implementer) using Ollama's tool calling API in Python.

Ollama's Python SDK natively parses Python functions—using their type hints and Google-style docstrings—into JSON tool schemas. To let an agent call multiple tools in sequence (e.g., searching the web twice or writing multiple files), you wrap ollama.chat in a while loop that executes response.message.tool_calls and feeds the outputs back with role="tool" until the model stops requesting tools.

1. Install Dependencies

Note that the duckduckgo-search package was renamed to ddgs:

pip install -U ollama pydantic ddgs

2. Define Safe, Typed Tools (tools.py)

Ollama relies heavily on parameter type annotations and docstrings to know when and how to call a function. Also, always sandbox file-writing tools to a dedicated ./workspace folder so a local LLM cannot accidentally overwrite your system files.

import os
from pathlib import Path
from ddgs import DDGS

WORKSPACE_DIR = Path("./workspace").resolve()
WORKSPACE_DIR.mkdir(parents=True, exist_ok=True)


def web_search(query: str, max_results: int = 5) -> str:
    """Search the web using DuckDuckGo for up-to-date facts, documentation, or news.

    Args:
        query: The search keywords or question to look up.
        max_results: Maximum number of search results to return (default 5).

    Returns:
        str: Formatted titles, URLs, and snippets from the search results.
    """
    print(f"    🌐 [Tool: web_search] Querying: '{query}'")
    try:
        results = DDGS().text(query, max_results=max_results)
        if not results:
            return "No search results found."
        formatted = [
            f"- {r.get('title')}: {r.get('body')} (Source: {r.get('href')})"
            for r in results
        ]
        return "\n".join(formatted)
    except Exception as e:
        return f"Search error: {e}"


def write_file(filepath: str, content: str) -> str:
    """Write or overwrite a file inside the project workspace directory.

    Args:
        filepath: Relative path inside the workspace (e.g., 'middleware/rate_limiter.py').
        content: Complete text or source code to write into the file.

    Returns:
        str: Confirmation message with the saved file path or an error.
    """
    print(f"    💾 [Tool: write_file] Writing: '{filepath}'")
    try:
        # Prevent path traversal outside ./workspace
        target_path = (WORKSPACE_DIR / filepath).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)):
            return "Security Error: Attempted to write outside workspace directory."

        target_path.parent.mkdir(parents=True, exist_ok=True)
        target_path.write_text(content, encoding="utf-8")
        return f"Successfully wrote {len(content)} characters to workspace/{filepath}"
    except Exception as e:
        return f"File write error: {e}"


def read_file(filepath: str) -> str:
    """Read an existing file from the project workspace directory.

    Args:
        filepath: Relative path inside the workspace to read.

    Returns:
        str: The contents of the file or an error message.
    """
    print(f"    📂 [Tool: read_file] Reading: '{filepath}'")
    try:
        target_path = (WORKSPACE_DIR / filepath).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)) or not target_path.exists():
            return f"Error: File '{filepath}' does not exist in workspace."
        return target_path.read_text(encoding="utf-8")
    except Exception as e:
        return f"File read error: {e}"

3. Upgrade run_llm with a Multi-Turn Tool Loop

Update the run_llm helper from your main script so it accepts an optional tools list. When tools are provided, it loops until the agent finishes invoking tools and returns its final text synthesis:

Python

from typing import Callable, List, Optional
from ollama import chat


def run_llm(
    model: str,
    system_prompt: str,
    user_prompt: str,
    schema=None,
    tools: Optional[List[Callable]] = None,
    max_tool_turns: int = 5,
):
    """Calls local Ollama with either Structured Outputs OR multi-turn Tool Calling."""
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt},
    ]

    # Map function names to actual callables for execution
    tool_map = {fn.__name__: fn for fn in tools} if tools else {}

    for _ in range(max_tool_turns):
        kwargs = {
            "model": model,
            "messages": messages,
            "options": {"temperature": 0.0},
        }
        if schema:
            kwargs["format"] = schema.model_json_schema()
        if tools:
            kwargs["tools"] = tools  # Pass Python functions directly

        response = chat(**kwargs)
        messages.append(response.message)  # Must append assistant message to history

        # If the model did not request any tools, return the final response
        if not response.message.tool_calls:
            content = response.message.content
            return schema.model_validate_json(content) if schema else content

        # Execute every tool call requested in this turn
        for tool_call in response.message.tool_calls:
            fn_name = tool_call.function.name
            fn_args = tool_call.function.arguments
            fn = tool_map.get(fn_name)

            if fn:
                result = fn(**fn_args)
            else:
                result = f"Error: Unknown tool '{fn_name}'"

            # Feed tool output back to the model
            messages.append({
                "role": "tool",
                "tool_name": fn_name,
                "content": str(result),
            })

    # Fallback if max_tool_turns was hit: force a final summary without tools
    final = chat(model=model, messages=messages, options={"temperature": 0.0})
    return final.message.content

4. Wire Tools into Finder and Implementer

Now attach web_search to Finder and write_file / read_file to Implementer (and optionally read_file to the Execution & Risk Critic so it can inspect the actual files written to disk):

Python

from tools import web_search, write_file, read_file


def finder_agent(state: SharedState, instructions: str) -> str:
    system_prompt = """You are the Finder. Use the `web_search` tool to gather accurate,
    current facts, library versions, and best practices. Cite your URLs.
    Do NOT interpret or recommend solutions."""
    prompt = f"Goal: {state.user_goal}\nInstructions: {instructions}"
    return run_llm(REASONING_MODEL, system_prompt, prompt, tools=[web_search])


def implementer_agent(state: SharedState, instructions: str, feedback: str = "") -> str:
    system_prompt = """You are the Implementer. Execute the Recommender's chosen strategy.
    Use `write_file` to save complete, production-ready code and config files to disk.
    After writing the files, return a summary of every file created and how they connect."""
    prompt = f"Goal: {state.user_goal}\nInstructions: {instructions}\nApproved Strategy:\n{state.recommendations}"
    if feedback:
        prompt += f"\n\n[CRITICAL REVISION REQUIRED - FIX THESE ISSUES]:\n{feedback}"
    return run_llm(
        REASONING_MODEL,
        system_prompt,
        prompt,
        tools=[write_file, read_file],
    )

Important Gotchas with Ollama Tool Calling

1. Use REASONING_MODEL (qwen3:14b) for tool-using agents: Smaller models (like 8b) sometimes hallucinate tool arguments or forget to stop calling tools in a loop. qwen3:14b or llama3.3 handles multi-turn tool calling reliably.

2. Don't combine format=<schema> and tools=[...] in the same call: Many local Ollama models struggle to emit valid tool-call tokens when constrained by a rigid JSON response schema simultaneously. Keep schema for routing/critique agents and tools for action agents.

Add a safe subprocess code-execution tool so the Implementer and Execution Critic can run pytest and verify the generated Python files in ./workspace before approving.

To let the Implementer and Execution & Risk Critic verify generated code without giving an LLM arbitrary shell access (rm -rf, network exfiltration, or infinite while True loops), use an allowlisted command runner locked to ./workspace with strict timeouts, output truncation, and stripped environment variables.

Because Ollama models shouldn't combine tools=[...] and format=<PydanticSchema> in the same turn, the Execution Critic uses a two-step pattern:

1. Inspection Phase (with tools): Uses read_file and run_workspace_tests to read the code and run pytest.

2. Verdict Phase (with Pydantic schema): Emits the strict CriticVerdict JSON based on the actual test logs.

1. Add the Sandboxed Execution Tool (tools.py)

Append run_workspace_tests and list_workspace_files to tools.py. Instead of passing shell=True, pass an explicit argument list to subprocess.run and allow only pytest or python scripts inside ./workspace:

import os
import subprocess
import sys
from pathlib import Path

WORKSPACE_DIR = Path("./workspace").resolve()
WORKSPACE_DIR.mkdir(parents=True, exist_ok=True)


def list_workspace_files() -> str:
    """List all files currently saved inside the project workspace directory.

    Returns:
        str: Newline-separated list of relative file paths in ./workspace.
    """
    files = [
        str(p.relative_to(WORKSPACE_DIR))
        for p in WORKSPACE_DIR.rglob("*")
        if p.is_file() and "__pycache__" not in p.parts and ".pytest_cache" not in p.parts
    ]
    return "\n".join(sorted(files)) if files else "Workspace is currently empty."


def run_workspace_tests(target: str = "") -> str:
    """Run pytest on the ./workspace directory or a specific test file inside it.

    Args:
        target: Optional relative path to a test file inside ./workspace
                (e.g., 'tests/test_rate_limiter.py'). Leave empty to run all tests.

    Returns:
        str: Exit code, stdout, and stderr from pytest (truncated to 4,000 chars).
    """
    print(f"    🧪 [Tool: run_workspace_tests] Running pytest on: '{target or 'all tests'}'")

    # 1. Validate target path stays strictly inside ./workspace
    if target:
        target_path = (WORKSPACE_DIR / target).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)) or not target_path.exists():
            return f"Security/Path Error: Target '{target}' does not exist inside ./workspace."
        cmd = [sys.executable, "-m", "pytest", "-v", str(target_path)]
    else:
        cmd = [sys.executable, "-m", "pytest", "-v"]

    # 2. Strip sensitive host environment variables (API keys, tokens, etc.)
    safe_env = {
        "PATH": os.environ.get("PATH", ""),
        "PYTHONPATH": str(WORKSPACE_DIR),
        "PYTHONDONTWRITEBYTECODE": "1",
    }

    try:
        # 3. Run without shell=True and enforce a strict 15-second timeout
        result = subprocess.run(
            cmd,
            cwd=str(WORKSPACE_DIR),
            env=safe_env,
            capture_output=True,
            text=True,
            timeout=15,
            shell=False,
        )
        output = (
            f"EXIT CODE: {result.returncode} ({'PASSED' if result.returncode == 0 else 'FAILED'})\n"
            f"--- STDOUT ---\n{result.stdout}\n"
            f"--- STDERR ---\n{result.stderr}"
        )
        # 4. Truncate huge stack traces so they don't blow up local LLM context windows
        if len(output) > 4000:
            output = output[:2000] + "\n... [TRUNCATED] ...\n" + output[-2000:]
        return output

    except subprocess.TimeoutExpired:
        return "EXECUTION ERROR: pytest timed out after 15 seconds (possible infinite loop or blocking call)."
    except Exception as e:
        return f"EXECUTION ERROR: {e}"

2. Update Implementer to Write & Self-Test Code

Give implementer_agent access to run_workspace_tests and list_workspace_files alongside write_file and read_file. Instruct it to write a pytest suite and fix any test failures before handing off to the Critic:

from tools import (
    web_search,
    write_file,
    read_file,
    list_workspace_files,
    run_workspace_tests,
)


def implementer_agent(state: SharedState, instructions: str, feedback: str = "") -> str:
    system_prompt = """You are the Implementer. Execute the Recommender's chosen strategy completely.
    Workflow:
    1. Use `write_file` to save clean, production-ready source files inside ./workspace.
    2. Write comprehensive unit tests (e.g., `tests/test_*.py`) covering normal behavior and edge cases.
    3. Run `run_workspace_tests()` to verify your code passes pytest.
    4. If any test fails, inspect the error, fix the code with `write_file`, and re-run `run_workspace_tests()` until EXIT CODE is 0.
    5. Return a concise summary of all created files and the final pytest results."""

    prompt = (
        f"Goal: {state.user_goal}\n"
        f"Instructions: {instructions}\n"
        f"Approved Strategy:\n{state.recommendations}"
    )
    if feedback:
        prompt += f"\n\n[CRITICAL REVISION REQUIRED - FIX THESE ISSUES]:\n{feedback}"

    return run_llm(
        REASONING_MODEL,
        system_prompt,
        prompt,
        tools=[write_file, read_file, list_workspace_files, run_workspace_tests],
        max_tool_turns=8,  # Allow enough turns to write code, write tests, fail, fix, and re-test
    )

3. Upgrade Critic to Independently Verify Workspace Files & Tests

Update critic_agent so that when critic_type == "Execution & Risk" and workspace files exist, it first inspects the actual files and runs pytest itself (preventing the Implementer from claiming tests passed when they didn't), and then outputs the structured CriticVerdict:

def critic_agent(
    critic_type: Literal["Logic & Fact", "Execution & Risk"],
    state: SharedState,
    artifact_to_review: str,
) -> CriticVerdict:
    if critic_type == "Logic & Fact":
        system_prompt = """You are the Logic & Fact Critic. Ruthlessly evaluate the output for
        hallucinations, unsupported leaps in logic, and data contradictions.
        Only mark APPROVED if there are zero HIGH severity flaws. Output strict JSON."""
        prompt = f"Original Goal: {state.user_goal}\n\nOutput to Review:\n{artifact_to_review}"
        return run_llm(REASONING_MODEL, system_prompt, prompt, schema=CriticVerdict)

    # --- EXECUTION & RISK CRITIC (Two-Phase: Tool Verification -> Structured Verdict) ---
    workspace_inventory = list_workspace_files()
    verification_report = "No files in workspace yet (evaluating strategy phase)."

    if workspace_inventory != "Workspace is currently empty.":
        print("    🕵️ [Execution Critic] Inspecting workspace files & running independent pytest...")
        verify_system = """You are a Senior QA & Security Engineer auditing code in ./workspace.
        1. Call `list_workspace_files()` and `read_file()` to inspect the actual implementation and tests.
        2. Call `run_workspace_tests()` to independently verify that all tests pass.
        3. Check for missing edge cases, security issues (e.g., hardcoded secrets, injection), or stubbed/fake tests.
        Summarize your objective audit findings."""

        verify_prompt = (
            f"User Goal: {state.user_goal}\n"
            f"Implementer Summary:\n{artifact_to_review}\n\n"
            "Inspect the files in ./workspace and run pytest now."
        )
        verification_report = run_llm(
            REASONING_MODEL,
            verify_system,
            verify_prompt,
            tools=[list_workspace_files, read_file, run_workspace_tests],
            max_tool_turns=6,
        )

    # Phase 2: Convert audit findings into strict CriticVerdict JSON
    verdict_system = """You are the Execution & Risk Critic. Based on the artifact and the
    independent workspace test audit, issue a strict JSON verdict.
    Rules:
    - If pytest failed (EXIT CODE != 0) or tests are trivial/missing, you MUST set status='REJECTED' and severity='HIGH'.
    - Only set status='APPROVED' when all tests pass and zero HIGH severity flaws remain."""

    verdict_prompt = (
        f"Original Goal: {state.user_goal}\n\n"
        f"Agent Output:\n{artifact_to_review}\n\n"
        f"Independent Audit & Pytest Results:\n{verification_report}"
    )
    return run_llm(REASONING_MODEL, verdict_system, verdict_prompt, schema=CriticVerdict)

Security & Stability Guardrails Built In

• shell=False + Explicit Command List: The LLM only controls the target relative file path—never raw shell flags or arbitrary binaries.

• Environment Stripping (safe_env): Even if you add OPENAI_API_KEY or database credentials to your host environment later, the subprocess running LLM-generated Python code cannot read os.environ secrets.

• 15-Second Timeout: Catches accidental infinite loops (while True:) in generated code before they hang your pipeline.

• Context Window Protection: Truncates pytest output to 4,000 characters so a massive traceback doesn't overflow your local model's KV cache.

(Note: Ensure pytest is installed in the same Python virtual environment running your agent script via pip install pytest.)

Convert this 7-agent Ollama architecture with DuckDuckGo search, file tools, and pytest execution into a stateful LangGraph workflow.

Porting your custom while loop into LangGraph gives you declarative state management, automatic tool-execution subgraphs (create_react_agent), native checkpointing (so you can inspect or resume runs), and clean conditional edges for the Critic rejection loops.

1. Install LangGraph & LangChain Ollama

pip install -U langgraph langchain-ollama langchain-core pydantic ddgs pytest

2. Wrap Tools with @tool (langgraph_tools.py)

LangGraph's prebuilt ReAct agents expect LangChain @tool-decorated functions. Keep the exact same workspace sandboxing and pytest guardrails:

import os
import subprocess
import sys
from pathlib import Path
from ddgs import DDGS
from langchain_core.tools import tool

WORKSPACE_DIR = Path("./workspace").resolve()
WORKSPACE_DIR.mkdir(parents=True, exist_ok=True)


@tool
def web_search(query: str, max_results: int = 5) -> str:
    """Search the web using DuckDuckGo for up-to-date facts, documentation, or news."""
    print(f"    🌐 [Tool: web_search] Querying: '{query}'")
    try:
        results = DDGS().text(query, max_results=max_results)
        if not results:
            return "No search results found."
        return "\n".join(
            f"- {r.get('title')}: {r.get('body')} (Source: {r.get('href')})"
            for r in results
        )
    except Exception as e:
        return f"Search error: {e}"


@tool
def write_file(filepath: str, content: str) -> str:
    """Write or overwrite a file inside the ./workspace directory."""
    print(f"    💾 [Tool: write_file] Writing: '{filepath}'")
    try:
        target_path = (WORKSPACE_DIR / filepath).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)):
            return "Security Error: Path traversal outside ./workspace is blocked."
        target_path.parent.mkdir(parents=True, exist_ok=True)
        target_path.write_text(content, encoding="utf-8")
        return f"Successfully wrote {len(content)} chars to workspace/{filepath}"
    except Exception as e:
        return f"File write error: {e}"


@tool
def read_file(filepath: str) -> str:
    """Read an existing file from the ./workspace directory."""
    print(f"    📂 [Tool: read_file] Reading: '{filepath}'")
    try:
        target_path = (WORKSPACE_DIR / filepath).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)) or not target_path.exists():
            return f"Error: File '{filepath}' does not exist in ./workspace."
        return target_path.read_text(encoding="utf-8")
    except Exception as e:
        return f"File read error: {e}"


@tool
def list_workspace_files() -> str:
    """List all files currently saved inside the ./workspace directory."""
    files = [
        str(p.relative_to(WORKSPACE_DIR))
        for p in WORKSPACE_DIR.rglob("*")
        if p.is_file() and "__pycache__" not in p.parts and ".pytest_cache" not in p.parts
    ]
    return "\n".join(sorted(files)) if files else "Workspace is currently empty."


@tool
def run_workspace_tests(target: str = "") -> str:
    """Run pytest on ./workspace or a specific test file inside it."""
    print(f"    🧪 [Tool: run_workspace_tests] Running pytest on: '{target or 'all tests'}'")
    if target:
        target_path = (WORKSPACE_DIR / target).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)) or not target_path.exists():
            return f"Path Error: Target '{target}' does not exist in ./workspace."
        cmd = [sys.executable, "-m", "pytest", "-v", str(target_path)]
    else:
        cmd = [sys.executable, "-m", "pytest", "-v"]

    safe_env = {
        "PATH": os.environ.get("PATH", ""),
        "PYTHONPATH": str(WORKSPACE_DIR),
        "PYTHONDONTWRITEBYTECODE": "1",
    }

    try:
        result = subprocess.run(
            cmd,
            cwd=str(WORKSPACE_DIR),
            env=safe_env,
            capture_output=True,
            text=True,
            timeout=15,
            shell=False,
        )
        output = (
            f"EXIT CODE: {result.returncode} ({'PASSED' if result.returncode == 0 else 'FAILED'})\n"
            f"--- STDOUT ---\n{result.stdout}\n"
            f"--- STDERR ---\n{result.stderr}"
        )
        return output[:2000] + "\n...[TRUNCATED]...\n" + output[-2000:] if len(output) > 4000 else output
    except subprocess.TimeoutExpired:
        return "EXECUTION ERROR: pytest timed out after 15 seconds."
    except Exception as e:
        return f"EXECUTION ERROR: {e}"

3. Complete LangGraph Workflow (langgraph_team.py)

In LangGraph, each agent is a node that reads and mutates a TypedDict state. Tool-using agents (Finder, Implementer, and Execution Critic's audit step) use create_react_agent to handle the multi-turn tool-calling loop automatically, while Manager and the Critic verdict steps use .with_structured_output(PydanticModel).

from typing import Literal, List, TypedDict
from pydantic import BaseModel, Field
from langchain_ollama import ChatOllama
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

from langgraph_tools import (
    web_search,
    write_file,
    read_file,
    list_workspace_files,
    run_workspace_tests,
)

# ==========================================
# 1. MODELS & STRUCTURED SCHEMAS
# ==========================================
reasoning_llm = ChatOllama(model="qwen3:14b", temperature=0.0)
fast_llm = ChatOllama(model="qwen3:8b", temperature=0.0)
MAX_RETRIES = 2


class ManagerDecision(BaseModel):
    next_agent: Literal[
        "finder", "analyzer", "recommender", "implementer", "next_steps", "DONE"
    ]
    task_instructions: str = Field(description="Specific instructions for the chosen agent.")
    reasoning: str = Field(description="Why this agent should run next.")


class CriticVerdict(BaseModel):
    status: Literal["APPROVED", "REJECTED"]
    severity: Literal["LOW", "HIGH"]
    critique: List[str] = Field(description="Specific flaws, hallucinations, or test failures.")
    required_fixes: str = Field(description="Exact instructions on what must change.")


# ==========================================
# 2. LANGGRAPH SHARED STATE
# ==========================================
class TeamState(TypedDict):
    user_goal: str
    next_agent: str
    task_instructions: str
    finder_data: str
    analyzer_report: str
    recommendations: str
    implementation: str
    next_steps: str
    # Critic tracking
    active_stage: str  # Tracks which agent is being reviewed
    critic_status: str
    critic_feedback: str
    retry_count: int


# ==========================================
# 3. PREBUILT TOOL-CALLING SUB-AGENTS
# ==========================================
finder_executor = create_react_agent(
    reasoning_llm,
    tools=[web_search],
    prompt=(
        "You are the Finder. Use `web_search` to gather accurate, up-to-date facts, "
        "library versions, and constraints. Cite URLs. Do NOT propose solutions."
    ),
)

implementer_executor = create_react_agent(
    reasoning_llm,
    tools=[write_file, read_file, list_workspace_files, run_workspace_tests],
    prompt=(
        "You are the Implementer. Write complete, production-ready code into ./workspace "
        "using `write_file`, write unit tests in `tests/`, and run `run_workspace_tests` "
        "until EXIT CODE is 0. Return a summary of created files and test results."
    ),
)

qa_auditor_executor = create_react_agent(
    reasoning_llm,
    tools=[list_workspace_files, read_file, run_workspace_tests],
    prompt=(
        "You are a Senior QA & Security Auditor. Inspect ./workspace files using "
        "`list_workspace_files` and `read_file`, and run `run_workspace_tests` to "
        "independently verify that tests exist, are non-trivial, and pass."
    ),
)


# ==========================================
# 4. GRAPH NODES
# ==========================================
def manager_node(state: TeamState) -> dict:
    structured_manager = reasoning_llm.with_structured_output(ManagerDecision)
    prompt = f"""
    Inspect the current state and route to the next specialist:
    - 'finder' if finder_data is EMPTY
    - 'analyzer' if finder_data is POPULATED and analyzer_report is EMPTY
    - 'recommender' if analyzer_report is POPULATED and recommendations is EMPTY
    - 'implementer' if recommendations is POPULATED and implementation is EMPTY
    - 'next_steps' if implementation is POPULATED and next_steps is EMPTY
    - 'DONE' ONLY when next_steps is POPULATED

    USER GOAL: {state['user_goal']}
    STATE STATUS:
    - finder_data: {'POPULATED' if state.get('finder_data') else 'EMPTY'}
    - analyzer_report: {'POPULATED' if state.get('analyzer_report') else 'EMPTY'}
    - recommendations: {'POPULATED' if state.get('recommendations') else 'EMPTY'}
    - implementation: {'POPULATED' if state.get('implementation') else 'EMPTY'}
    - next_steps: {'POPULATED' if state.get('next_steps') else 'EMPTY'}
    """
    decision = structured_manager.invoke([HumanMessage(content=prompt)])
    print(f"\n🧠 [Manager] -> {decision.next_agent.upper()} ({decision.reasoning})")
    return {
        "next_agent": decision.next_agent,
        "task_instructions": decision.task_instructions,
        "critic_feedback": "",
        "retry_count": 0,
    }


def finder_node(state: TeamState) -> dict:
    msg = f"Goal: {state['user_goal']}\nInstructions: {state['task_instructions']}"
    res = finder_executor.invoke({"messages": [HumanMessage(content=msg)]})
    return {"finder_data": res["messages"][-1].content}


def analyzer_node(state: TeamState) -> dict:
    sys_msg = SystemMessage(
        content="You are the Analyzer. Synthesize Finder data into patterns, constraints, and root causes. Do NOT propose solutions."
    )
    user_msg = f"Goal: {state['user_goal']}\nInstructions: {state['task_instructions']}\nFinder Data:\n{state['finder_data']}"
    if state.get("critic_feedback"):
        user_msg += f"\n\n[REVISION REQUIRED]:\n{state['critic_feedback']}"

    res = reasoning_llm.invoke([sys_msg, HumanMessage(content=user_msg)])
    return {"analyzer_report": res.content, "active_stage": "analyzer"}


def recommender_node(state: TeamState) -> dict:
    sys_msg = SystemMessage(
        content="You are the Recommender. Present 2-3 options with pros/cons/costs, then declare ONE optimal strategy."
    )
    user_msg = f"Goal: {state['user_goal']}\nInstructions: {state['task_instructions']}\nAnalysis:\n{state['analyzer_report']}"
    if state.get("critic_feedback"):
        user_msg += f"\n\n[REVISION REQUIRED]:\n{state['critic_feedback']}"

    res = reasoning_llm.invoke([sys_msg, HumanMessage(content=user_msg)])
    return {"recommendations": res.content, "active_stage": "recommender"}


def implementer_node(state: TeamState) -> dict:
    user_msg = (
        f"Goal: {state['user_goal']}\n"
        f"Instructions: {state['task_instructions']}\n"
        f"Approved Strategy:\n{state['recommendations']}"
    )
    if state.get("critic_feedback"):
        user_msg += f"\n\n[REVISION REQUIRED - FIX THESE ISSUES]:\n{state['critic_feedback']}"

    res = implementer_executor.invoke(
        {"messages": [HumanMessage(content=user_msg)]},
        config={"recursion_limit": 20},
    )
    return {"implementation": res["messages"][-1].content, "active_stage": "implementer"}


def logic_critic_node(state: TeamState) -> dict:
    print("  🔍 [Logic & Fact Critic] Reviewing analysis...")
    structured_critic = reasoning_llm.with_structured_output(CriticVerdict)
    prompt = (
        "Evaluate the Analyzer report for hallucinations, unsupported leaps, or contradictions. "
        "Only approve if zero HIGH severity flaws exist.\n\n"
        f"Goal: {state['user_goal']}\nFinder Data:\n{state['finder_data']}\n"
        f"Analyzer Report:\n{state['analyzer_report']}"
    )
    verdict = structured_critic.invoke([HumanMessage(content=prompt)])
    return _format_critic_update("Logic & Fact", verdict, state)


def execution_critic_node(state: TeamState) -> dict:
    stage = state["active_stage"]
    print(f"  🔍 [Execution & Risk Critic] Reviewing {stage}...")
    structured_critic = reasoning_llm.with_structured_output(CriticVerdict)

    if stage == "recommender":
        prompt = (
            "Evaluate the Recommender's strategy for feasibility, edge cases, and constraint alignment.\n\n"
            f"Goal: {state['user_goal']}\nRecommendations:\n{state['recommendations']}"
        )
        verdict = structured_critic.invoke([HumanMessage(content=prompt)])
        return _format_critic_update("Execution & Risk", verdict, state)

    # Stage == "implementer": Run independent QA audit + pytest first
    audit_res = qa_auditor_executor.invoke(
        {
            "messages": [
                HumanMessage(
                    content=(
                        f"Goal: {state['user_goal']}\nImplementer Summary:\n{state['implementation']}\n"
                        "Inspect ./workspace files and run pytest now."
                    )
                )
            ]
        },
        config={"recursion_limit": 15},
    )
    audit_report = audit_res["messages"][-1].content

    verdict_prompt = (
        "Based on the Implementer summary and the independent QA/pytest audit below, issue a verdict. "
        "If pytest failed or tests are missing/trivial, set status='REJECTED' and severity='HIGH'.\n\n"
        f"Goal: {state['user_goal']}\nImplementer Summary:\n{state['implementation']}\n"
        f"Independent QA & Pytest Audit:\n{audit_report}"
    )
    verdict = structured_critic.invoke([HumanMessage(content=verdict_prompt)])
    return _format_critic_update("Execution & Risk", verdict, state)


def _format_critic_update(critic_name: str, verdict: CriticVerdict, state: TeamState) -> dict:
    if verdict.status == "APPROVED":
        print(f"  ✅ [{critic_name} Critic] APPROVED.")
        return {"critic_status": "APPROVED", "critic_feedback": ""}

    new_retries = state.get("retry_count", 0) + 1
    print(f"  ❌ [{critic_name} Critic] REJECTED (Attempt {new_retries}/{MAX_RETRIES}): {verdict.critique}")
    if new_retries > MAX_RETRIES:
        print("  ⚠️ Max retries reached. Force-approving to prevent infinite loop.")
        return {"critic_status": "APPROVED", "critic_feedback": ""}

    return {
        "critic_status": "REJECTED",
        "critic_feedback": f"Critique: {verdict.critique}\nRequired Fixes: {verdict.required_fixes}",
        "retry_count": new_retries,
    }


def next_steps_node(state: TeamState) -> dict:
    sys_msg = SystemMessage(
        content="You are the Next Steps Coordinator. Provide: (1) Summary, (2) Chronological Action Checklist, (3) 2-3 KPIs."
    )
    user_msg = f"Goal: {state['user_goal']}\nImplementation:\n{state['implementation']}"
    res = fast_llm.invoke([sys_msg, HumanMessage(content=user_msg)])
    return {"next_steps": res.content}


# ==========================================
# 5. CONDITIONAL EDGE ROUTERS
# ==========================================
def route_from_manager(state: TeamState) -> str:
    target = state["next_agent"]
    return END if target == "DONE" else target


def route_from_critic(state: TeamState) -> str:
    if state["critic_status"] == "APPROVED":
        return "manager"
    # Route back to whichever agent produced the rejected artifact
    return state["active_stage"]


# ==========================================
# 6. COMPILE & RUN GRAPH
# ==========================================
def build_graph():
    workflow = StateGraph(TeamState)

    # Add nodes
    workflow.add_node("manager", manager_node)
    workflow.add_node("finder", finder_node)
    workflow.add_node("analyzer", analyzer_node)
    workflow.add_node("recommender", recommender_node)
    workflow.add_node("implementer", implementer_node)
    workflow.add_node("logic_critic", logic_critic_node)
    workflow.add_node("execution_critic", execution_critic_node)
    workflow.add_node("next_steps", next_steps_node)

    # Entry point
    workflow.add_edge(START, "manager")

    # Manager hub routing
    workflow.add_conditional_edges(
        "manager",
        route_from_manager,
        {
            "finder": "finder",
            "analyzer": "analyzer",
            "recommender": "recommender",
            "implementer": "implementer",
            "next_steps": "next_steps",
            END: END,
        },
    )

    # Specialist -> Critic / Manager edges
    workflow.add_edge("finder", "manager")
    workflow.add_edge("analyzer", "logic_critic")
    workflow.add_edge("recommender", "execution_critic")
    workflow.add_edge("implementer", "execution_critic")
    workflow.add_edge("next_steps", "manager")

    # Critic conditional loops
    workflow.add_conditional_edges(
        "logic_critic",
        route_from_critic,
        {"manager": "manager", "analyzer": "analyzer"},
    )
    workflow.add_conditional_edges(
        "execution_critic",
        route_from_critic,
        {
            "manager": "manager",
            "recommender": "recommender",
            "implementer": "implementer",
        },
    )

    return workflow.compile(checkpointer=MemorySaver())


if __name__ == "__main__":
    app = build_graph()
    config = {"configurable": {"thread_id": "rate-limiter-run-1"}}

    initial_state: TeamState = {
        "user_goal": "Build a thread-safe token-bucket rate limiter in Python with a full pytest suite.",
        "next_agent": "",
        "task_instructions": "",
        "finder_data": "",
        "analyzer_report": "",
        "recommendations": "",
        "implementation": "",
        "next_steps": "",
        "active_stage": "",
        "critic_status": "",
        "critic_feedback": "",
        "retry_count": 0,
    }

    final_state = app.invoke(initial_state, config=config)

    print("\n" + "=" * 60)
    print("FINAL IMPLEMENTATION SUMMARY:\n", final_state["implementation"])
    print("\n" + "=" * 60)
    print("NEXT STEPS & KPIS:\n", final_state["next_steps"])

What LangGraph Adds to Your Architecture

1. Declarative Critic Loops (route_from_critic): Instead of manual for loops wrapping functions, execution_critic dynamically routes back to either recommender or implementer depending on state["active_stage"], and only advances to manager once critic_status == "APPROVED".

2. State Checkpointing (MemorySaver): Because thread_id is passed into app.invoke(), every node transition is saved. You can call app.get_state_history(config) after a run to inspect every rejected draft and critic verdict, or swap MemorySaver() for SqliteSaver to persist runs across restarts.

3. Subgraph Isolation: finder_executor, implementer_executor, and qa_auditor_executor keep their noisy internal tool-call message histories encapsulated inside create_react_agent, returning only their clean final summary into the top-level TeamState so the Manager and Analyzer contexts stay lean.

How does Pydantic factor in?

In a multi-agent system running on local LLMs (like Ollama), Pydantic acts as the deterministic "glue" between probabilistic models and strict Python control flow.

Without Pydantic, local models frequently wrap JSON in conversational filler ("Sure, here is the routing decision: ```json..."), misspell dictionary keys, or return invalid values that crash your orchestration loop with a KeyError or `JSONDecodeError`.

Pydantic solves this across four specific layers in the architecture we built:

1. Grammar-Constrained Generation (Eliminating Syntax Crashes)

When you pass ManagerDecision or CriticVerdict into Ollama's format= parameter (or LangChain's .with_structured_output()), Pydantic compiles your Python class into a JSON Schema:

class CriticVerdict(BaseModel):
    status: Literal["APPROVED", "REJECTED"]
    severity: Literal["LOW", "HIGH"]
    critique: List[str]
    required_fixes: str

Ollama uses this schema at the token-sampling level (grammar-based sampling) to physically mask out any tokens that would violate the JSON structure or allowed values. For example, when generating the status field, the model is mathematically prevented from outputting "Looks good!"—it can only emit "APPROVED" or "REJECTED".

2. Strict Type & Value Validation (Guardrails for Routing)

Using Literal[...] inside Pydantic models guarantees that your Python if/elif routers and LangGraph conditional edges never receive an unexpected string:

• Manager Routing: next_agent: Literal["finder", "analyzer", ...] ensures the Manager never invents a non-existent agent name like "researcher" or "coder".

• Runtime Parsing: Calling CriticVerdict.model_validate_json(content) converts raw text into a real Python object with full IDE autocomplete (verdict.status, verdict.critique) rather than fragile dictionary lookups (verdict["status"]).

3. In-Schema Prompt Engineering (Field(description=...))

Pydantic's Field(description="...") isn't just documentation for humans—it gets embedded directly into the JSON schema sent to the LLM:

required_fixes: str = Field(description="Exact instructions on what must change.")

This acts as micro-instructions right at the moment the model generates that specific field, keeping outputs concise and targeted without bloating your main system prompt.

4. Automatic Tool-Argument Parsing

When you define tools with Python type hints (def web_search(query: str, max_results: int = 5)), both the native ollama SDK and LangChain's @tool decorator use Pydantic under the hood to:

1. Convert your function signature into a tool schema for the LLM.

2. Coerce and validate the arguments the LLM generates before calling your Python function (e.g., ensuring max_results is cast to an int rather than a string "5").

Where NOT to Use Pydantic Schemas

Notice that we intentionally left Finder, Analyzer, Recommender, Implementer, and Next Steps as plain text/Markdown outputs rather than forcing them into Pydantic schemas:

• Reasoning Degradation: Forcing an LLM to write complex multi-file Python code or deep analytical essays inside a single escaped JSON string ("code": "def foo():\n return \"bar\"") wastes tokens on escape characters and measurably degrades local model reasoning quality.

• Best Practice: Use Pydantic for control-plane agents (Managers, Critics, Routers, Tool Arguments) and free-form Markdown for data-plane agents (Analysts, Writers, Coders).

Show me how to add custom Pydantic @model_validator rules to CriticVerdict so a REJECTED verdict is automatically rejected if required_fixes is empty.

In Pydantic v2, you use @model_validator(mode="after") to validate relationships across multiple fields once individual field types have been parsed.

While grammar-constrained sampling forces the local LLM to emit valid JSON keys (status, critique, required_fixes), it cannot stop a lazy model from outputting "status": "REJECTED" alongside "required_fixes": "" or "critique": []. A @model_validator catches this logical contradiction immediately and raises a ValidationError that you can feed back to the model for self-correction.

1. Define CriticVerdict with Cross-Field Rules

Here is a production-ready CriticVerdict enforcing three invariants:

1. If status == "REJECTED", critique must contain at least one non-empty item.

2. If status == "REJECTED", required_fixes must be substantive (not blank or placeholder text like "None" or "N/A").

3. If status == "APPROVED", severity cannot be "HIGH" (prevents contradictory approvals).

from typing import List, Literal
from pydantic import BaseModel, Field, model_validator


class CriticVerdict(BaseModel):
    status: Literal["APPROVED", "REJECTED"]
    severity: Literal["LOW", "HIGH"]
    critique: List[str] = Field(
        default_factory=list,
        description="Specific flaws, hallucinations, or test failures found.",
    )
    required_fixes: str = Field(
        default="",
        description="Actionable step-by-step instructions on what must change. Required if REJECTED.",
    )

    @model_validator(mode="after")
    def enforce_rejection_actionability(self) -> "CriticVerdict":
        # Strip whitespace to catch "   " or empty strings
        clean_fixes = self.required_fixes.strip()
        clean_critiques = [c.strip() for c in self.critique if c.strip()]

        if self.status == "REJECTED":
            if not clean_critiques:
                raise ValueError(
                    "Invalid CriticVerdict: When status is 'REJECTED', 'critique' must list at least one specific flaw."
                )
            if len(clean_fixes) < 15 or clean_fixes.lower() in {"none", "n/a", "fix it"}:
                raise ValueError(
                    "Invalid CriticVerdict: When status is 'REJECTED', 'required_fixes' must contain concrete, actionable instructions (at least 15 characters)."
                )

        if self.status == "APPROVED" and self.severity == "HIGH":
            raise ValueError(
                "Invalid CriticVerdict: Cannot set status='APPROVED' while severity='HIGH'."
            )

        # Store normalized values back on the model
        self.required_fixes = clean_fixes
        self.critique = clean_critiques
        return self

2. Self-Healing Retry Wrapper (Native Ollama & LangChain)

Raising a ValueError inside @model_validator is only useful if your runtime catches ValidationError and asks the local model to fix its schema mistake instead of crashing the pipeline.

Option A: Native ollama SDK Wrapper

from pydantic import ValidationError
from ollama import chat


def run_validated_llm(
    model: str,
    system_prompt: str,
    user_prompt: str,
    schema: type[BaseModel],
    max_validation_retries: int = 2,
):
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt},
    ]

    for attempt in range(max_validation_retries + 1):
        response = chat(
            model=model,
            messages=messages,
            format=schema.model_json_schema(),
            options={"temperature": 0.0},
        )
        raw_json = response.message.content

        try:
            # Triggers both type checks AND @model_validator rules
            return schema.model_validate_json(raw_json)
        except ValidationError as err:
            if attempt == max_validation_retries:
                raise
            print(f"    ⚠️ [Schema Validation Retry {attempt + 1}] Feeding error back to LLM...")
            messages.append({"role": "assistant", "content": raw_json})
            messages.append({
                "role": "user",
                "content": (
                    f"Your JSON failed validation rules:\n{err}\n\n"
                    "Regenerate the JSON strictly obeying all validation constraints."
                ),
            })

Option B: LangChain / LangGraph .with_retry()

In LangGraph, .with_structured_output(CriticVerdict) automatically runs your @model_validator during parsing. Chaining .with_retry() retries automatically if the validator raises a ValidationError:

from langchain_ollama import ChatOllama

reasoning_llm = ChatOllama(model="qwen3:14b", temperature=0.0)

# Automatically retries up to 3 times if @model_validator raises ValueError
structured_critic = reasoning_llm.with_structured_output(CriticVerdict).with_retry(
    stop_after_attempt=3
)

3. Quick Unit Test (test_verdict_validator.py)

You can verify the validator rules directly without calling an LLM:

import pytest
from pydantic import ValidationError


def test_rejected_without_fixes_raises_error():
    with pytest.raises(ValidationError, match="required_fixes"):
        CriticVerdict(
            status="REJECTED",
            severity="HIGH",
            critique=["Missing rate-limit headers"],
            required_fixes="   ",  # Blank string triggers validator failure
        )


def test_valid_rejection_passes():
    verdict = CriticVerdict(
        status="REJECTED",
        severity="HIGH",
        critique=["Missing thread lock around token refill"],
        required_fixes="Wrap refill() and consume() in a threading.Lock() context manager.",
    )
    assert verdict.status == "REJECTED"

Show me how to replace the LangGraph TypedDict TeamState with a Pydantic BaseModel for runtime state validation and default field values.

LangGraph natively accepts a Pydantic BaseModel as the schema passed to StateGraph(TeamState). Switching from TypedDict to BaseModel lets you declare default field values (eliminating boilerplate empty-string initialization dictionaries), enforce non-negative bounds (Field(ge=0)), and validate incoming inputs at graph entry.

1. Define TeamState as a Pydantic BaseModel

You can use Annotated reducers (like operator.add for an audit trail) directly on Pydantic fields alongside standard defaults and field validators:

import operator
from typing import Annotated, List, Literal
from pydantic import BaseModel, Field, field_validator


class TeamState(BaseModel):
    # Required input field — validated at graph entry
    user_goal: str = Field(min_length=5, description="The user's primary objective.")

    # Routing & orchestration fields with clean defaults
    next_agent: Literal[
        "", "finder", "analyzer", "recommender", "implementer", "next_steps", "DONE"
    ] = ""
    task_instructions: str = ""

    # Specialist artifacts
    finder_data: str = ""
    analyzer_report: str = ""
    recommendations: str = ""
    implementation: str = ""
    next_steps: str = ""

    # Critic loop tracking
    active_stage: Literal["", "analyzer", "recommender", "implementer"] = ""
    critic_status: Literal["", "APPROVED", "REJECTED"] = ""
    critic_feedback: str = ""
    retry_count: int = Field(default=0, ge=0)

    # Append-only audit log using a LangGraph reducer
    audit_log: Annotated[List[str], operator.add] = Field(default_factory=list)

    @field_validator("user_goal")
    @classmethod
    def strip_goal(cls, v: str) -> str:
        cleaned = v.strip()
        if len(cleaned) < 5:
            raise ValueError("user_goal must contain at least 5 non-whitespace characters.")
        return cleaned

2. Update Nodes to Use Attribute Access (state.field)

When StateGraph is initialized with a BaseModel, each node receives an actual TeamState instance as its first argument rather than a plain dict. Replace state["key"] and state.get("key") with dot notation (state.key), while still returning partial update dictionaries from each node:

from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


def manager_node(state: TeamState) -> dict:
    structured_manager = reasoning_llm.with_structured_output(ManagerDecision)
    prompt = f"""
    Inspect the current state and route to the next specialist:
    - 'finder' if finder_data is EMPTY
    - 'analyzer' if finder_data is POPULATED and analyzer_report is EMPTY
    - 'recommender' if analyzer_report is POPULATED and recommendations is EMPTY
    - 'implementer' if recommendations is POPULATED and implementation is EMPTY
    - 'next_steps' if implementation is POPULATED and next_steps is EMPTY
    - 'DONE' ONLY when next_steps is POPULATED

    USER GOAL: {state.user_goal}
    STATE STATUS:
    - finder_data: {'POPULATED' if state.finder_data else 'EMPTY'}
    - analyzer_report: {'POPULATED' if state.analyzer_report else 'EMPTY'}
    - recommendations: {'POPULATED' if state.recommendations else 'EMPTY'}
    - implementation: {'POPULATED' if state.implementation else 'EMPTY'}
    - next_steps: {'POPULATED' if state.next_steps else 'EMPTY'}
    """
    decision = structured_manager.invoke([HumanMessage(content=prompt)])
    return {
        "next_agent": decision.next_agent,
        "task_instructions": decision.task_instructions,
        "critic_feedback": "",
        "retry_count": 0,
        "audit_log": [f"Manager routed to {decision.next_agent}"],
    }


def analyzer_node(state: TeamState) -> dict:
    sys_msg = SystemMessage(
        content="You are the Analyzer. Synthesize Finder data into patterns, constraints, and root causes."
    )
    user_msg = (
        f"Goal: {state.user_goal}\n"
        f"Instructions: {state.task_instructions}\n"
        f"Finder Data:\n{state.finder_data}"
    )
    if state.critic_feedback:
        user_msg += f"\n\n[REVISION REQUIRED]:\n{state.critic_feedback}"

    res = reasoning_llm.invoke([sys_msg, HumanMessage(content=user_msg)])
    return {
        "analyzer_report": res.content,
        "active_stage": "analyzer",
        "audit_log": ["Analyzer generated report"],
    }


# Conditional routers also use attribute access
def route_from_manager(state: TeamState) -> str:
    return END if state.next_agent == "DONE" else state.next_agent


def route_from_critic(state: TeamState) -> str:
    return "manager" if state.critic_status == "APPROVED" else state.active_stage

3. Invoke the Graph with Clean Defaults

Because every field except user_goal has a default value in TeamState, you no longer need to pass 11 empty fields when starting a run—and you can pass either a minimal dictionary or a TeamState instance:

workflow = StateGraph(TeamState)
# ... (add_node and add_edge calls remain identical) ...
app = workflow.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "rate-limiter-run-1"}}

# Option A: Pass a minimal dict (Pydantic populates defaults & runs @field_validator)
final_state = app.invoke({"user_goal": "Build a thread-safe token-bucket rate limiter."}, config=config)

# Option B: Pass a validated TeamState instance directly
initial = TeamState(user_goal="Build a thread-safe token-bucket rate limiter.")
final_state = app.invoke(initial, config=config)

# IMPORTANT: LangGraph returns the final state as a dict, not a BaseModel instance
print("Implementation:", final_state["implementation"])
print("Audit trail:", final_state["audit_log"])

Two Important LangGraph + Pydantic Gotchas

1. Input validation vs. internal node updates: LangGraph runs full Pydantic validation on the initial input entering the first node, and passes a TeamState instance into every node. However, partial dictionaries returned between internal nodes are merged into state without re-running top-level @model_validator hooks—so keep strict validation inside node logic or structured output schemas (CriticVerdict, ManagerDecision).

2. Graph output type: Even when StateGraph(TeamState) uses a BaseModel, app.invoke(...) returns a plain Python dict. If you want a TeamState object at the very end, re-hydrate it with TeamState.model_validate(final_state).

Add a ranking agent. For example: rate a recommendation on feasibility, practicality, benefit, complexity, risk

Adding a Ranker Agent turns the Recommender's qualitative prose into a quantitative decision matrix before the Implementer writes a single line of code.

By using a Pydantic schema with bounded integer scores (1–10) and a @computed_field for the weighted score, Python handles all the arithmetic deterministically—preventing local LLMs from making basic addition errors when ranking options.

1. Updated Workflow Topology

Place the Ranker immediately after the Recommender (and before the Execution & Risk Critic), so the Critic can verify that the highest-scoring option actually aligns with the user's constraints:

[Recommender] ──► [Ranker (Scores Options)] ──► [Execution Critic] ──► [Manager] ──► [Implementer]

2. Pydantic Scoring Schema (RankingMatrix)

Define a structured schema where each option is scored across your five dimensions (feasibility, practicality, benefit, complexity, risk). Because complexity and risk are negative factors, invert them in the @computed_field formula so higher total scores always win:

from typing import List, Literal
from pydantic import BaseModel, Field, computed_field, model_validator


class RankedOption(BaseModel):
    option_name: str = Field(description="Short title of the proposed option.")
    feasibility: int = Field(ge=1, le=10, description="1-10: Technical & resource viability (10 = easiest to build).")
    practicality: int = Field(ge=1, le=10, description="1-10: Operational ease & maintainability (10 = most practical).")
    benefit: int = Field(ge=1, le=10, description="1-10: Impact & value delivered toward the goal (10 = highest impact).")
    complexity: int = Field(ge=1, le=10, description="1-10: Architectural & code complexity (10 = most complex / hardest).")
    risk: int = Field(ge=1, le=10, description="1-10: Security, performance, or failure risk (10 = highest risk).")
    rationale: str = Field(description="Concise 1-2 sentence justification for these scores.")

    @computed_field
    @property
    def net_score(self) -> float:
        """Weighted score: rewards benefit/feasibility/practicality, penalizes complexity/risk."""
        positive = (self.benefit * 1.5) + (self.feasibility * 1.2) + (self.practicality * 1.2)
        penalty = (self.complexity * 0.8) + (self.risk * 1.0)
        return round(positive - penalty, 2)


class RankingMatrix(BaseModel):
    evaluated_options: List[RankedOption] = Field(min_length=1)
    winning_option: str = Field(description="Exact option_name of the highest-scoring option.")
    selection_summary: str = Field(description="Why the winning option is best for the Implementer to build.")

    @model_validator(mode="after")
 

3. Update TeamState and Add ranker_node

Add ranking_report: str = "" to your Pydantic TeamState, update ManagerDecision to include "ranker", and define ranker_node:

class ManagerDecision(BaseModel):
    next_agent: Literal[
        "finder", "analyzer", "recommender", "ranker", "implementer", "next_steps", "DONE"
    ]
    task_instructions: str = Field(description="Specific instructions for the chosen agent.")
    reasoning: str = Field(description="Why this agent should run next.")


class TeamState(BaseModel):
    user_goal: str = Field(min_length=5)
    next_agent: Literal[
        "", "finder", "analyzer", "recommender", "ranker", "implementer", "next_steps", "DONE"
    ] = ""
    task_instructions: str = ""
    finder_data: str = ""
    analyzer_report: str = ""
    recommendations: str = ""
    ranking_report: str = ""  # <-- New field storing the formatted scorecard
    implementation: str = ""
    next_steps: str = ""
    active_stage: Literal["", "analyzer", "recommender", "ranker", "implementer"] = ""
    critic_status: Literal["", "APPROVED", "REJECTED"] = ""
    critic_feedback: str = ""
    retry_count: int = Field(default=0, ge=0)
    audit_log: Annotated[List[str], operator.add] = Field(default_factory=list)

Now implement ranker_node using .with_structured_output(RankingMatrix) and format its output into a clean Markdown scorecard table for downstream agents (Execution Critic and Implementer):

def ranker_node(state: TeamState) -> dict:
    print("  📊 [Ranker] Scoring strategic options...")
    structured_ranker = reasoning_llm.with_structured_output(RankingMatrix).with_retry(
        stop_after_attempt=3
    )

    prompt = f"""
    Score EVERY option proposed by the Recommender on a 1-10 scale across:
    - Feasibility (10 = easiest to build with current stack)
    - Practicality (10 = easiest to maintain and operate)
    - Benefit (10 = highest value toward the user goal)
    - Complexity (10 = most complex/over-engineered)
    - Risk (10 = highest failure, security, or edge-case risk)

    USER GOAL: {state.user_goal}
    CONSTRAINTS & ANALYSIS:\n{state.analyzer_report}
    PROPOSED OPTIONS:\n{state.recommendations}
    """
    if state.critic_feedback:
        prompt += f"\n\n[REVISION REQUIRED]:\n{state.critic_feedback}"

    matrix: RankingMatrix = structured_ranker.invoke([HumanMessage(content=prompt)])

    # Format as a scannable Markdown table for the Critic and Implementer
    table_lines = [
        f"### 🏆 Selected Strategy: **{matrix.winning_option}**",
        f"{matrix.selection_summary}\n",
        "| Rank | Option | Feasibility | Practicality | Benefit | Complexity | Risk | Net Score |",
        "| :--- | :--- | :---: | :---: | :---: | :---: | :---: | :---: |",
    ]
    for idx, opt in enumerate(matrix.evaluated_options, start=1):
        table_lines.append(
            f"| #{idx} | **{opt.option_name}** | {opt.feasibility}/10 | {opt.practicality}/10 | "
            f"{opt.benefit}/10 | {opt.complexity}/10 | {opt.risk}/10 | **{opt.net_score}** |"
        )
        table_lines.append(f"> *{opt.option_name} rationale:* {opt.rationale}")

    formatted_report = "\n".join(table_lines)

    return {
        "ranking_report": formatted_report,
        "active_stage": "ranker",
        "audit_log": [f"Ranker selected '{matrix.winning_option}' (Score: {matrix.evaluated_options[0].net_score})"],
    }

4. Wire the Ranker into the LangGraph Edges & Implementer

Update three places in your graph wiring so the Ranker sits cleanly between recommender and execution_critic:

# 1. Register node
workflow.add_node("ranker", ranker_node)

# 2. Update Manager's conditional routing map
workflow.add_conditional_edges(
    "manager",
    route_from_manager,
    {
        "finder": "finder",
        "analyzer": "analyzer",
        "recommender": "recommender",
        "ranker": "ranker",
        "implementer": "implementer",
        "next_steps": "next_steps",
        END: END,
    },
)

# 3. Route Recommender -> Ranker -> Execution Critic
workflow.add_edge("recommender", "ranker")
workflow.add_edge("ranker", "execution_critic")

# 4. Allow Execution Critic to send rejected rankings back to Ranker
workflow.add_conditional_edges(
    "execution_critic",
    route_from_critic,
    {
        "manager": "manager",
        "recommender": "recommender",
        "ranker": "ranker",
        "implementer": "implementer",
    },
)

Finally, update manager_node's routing rules ('ranker' if recommendations is POPULATED and ranking_report is EMPTY) and pass {state.ranking_report} into implementer_node so the Implementer builds only the winning option rather than guessing between the Recommender's 2–3 choices.

Show me how to let the Manager dynamically set the scoring weights (feasibility, practicality, benefit, complexity, risk) in TeamState based on the user's goal before the Ranker scores the options.

To let the Manager dynamically tailor the Ranker's priorities (for example, cranking up risk and practicality weights for a production fintech system, or boosting feasibility and benefit for a weekend hackathon prototype), define a ScoringWeights Pydantic model stored directly in TeamState and inject those weights into RankingMatrix via Pydantic's validation context (model_validate(..., context=...)).

Using ValidationInfo context keeps RankedOption clean while still computing every weighted net_score deterministically in Python.

1. Define ScoringWeights & Update TeamState

Create a bounded ScoringWeights model (0.5 to 2.5 multipliers) so the Manager cannot accidentally zero out a dimension or pass negative numbers that invert the math:

from typing import Annotated, List, Literal, Optional
import operator
from pydantic import BaseModel, Field, ValidationInfo, model_validator


class ScoringWeights(BaseModel):
    feasibility: float = Field(default=1.2, ge=0.5, le=2.5, description="Weight for technical/resource viability.")
    practicality: float = Field(default=1.2, ge=0.5, le=2.5, description="Weight for operational ease & maintainability.")
    benefit: float = Field(default=1.5, ge=0.5, le=2.5, description="Weight for business/user value delivered.")
    complexity: float = Field(default=0.8, ge=0.5, le=2.5, description="Penalty weight for architectural complexity.")
    risk: float = Field(default=1.0, ge=0.5, le=2.5, description="Penalty weight for security, reliability, or failure risk.")
    weight_rationale: str = Field(
        default="Standard balanced weighting.",
        description="1-sentence explanation of why these weights fit the user's goal.",
    )


class ManagerDecision(BaseModel):
    next_agent: Literal[
        "finder", "analyzer", "recommender", "ranker", "implementer", "next_steps", "DONE"
    ]
    task_instructions: str = Field(description="Specific instructions for the chosen agent.")
    reasoning: str = Field(description="Why this agent should run next.")
    # Populated by Manager when routing to 'ranker' (or during initial planning)
    scoring_weights: Optional[ScoringWeights] = Field(
        default=None,
        description="Required when next_agent == 'ranker'. Tailor weights to the user's goal and constraints.",
    )

    @model_validator(mode="after")
    def require_weights_for_ranker(self) -> "ManagerDecision":
        if self.next_agent == "ranker" and self.scoring_weights is None:
            raise ValueError(
                "Invalid ManagerDecision: You must provide 'scoring_weights' when routing to 'ranker'."
            )
        return self


class TeamState(BaseModel):
    user_goal: str = Field(min_length=5)
    next_agent: Literal[
        "", "finder", "analyzer", "recommender", "ranker", "implementer", "next_steps", "DONE"
    ] = ""
    task_instructions: str = ""
    finder_data: str = ""
    analyzer_report: str = ""
    recommendations: str = ""
    scoring_weights: ScoringWeights = Field(default_factory=ScoringWeights)  # <-- Stored in state
    ranking_report: str = ""
    implementation: str = ""
    next_steps: str = ""
    active_stage: Literal["", "analyzer", "recommender", "ranker", "implementer"] = ""
    critic_status: Literal["", "APPROVED", "REJECTED"] = ""
    critic_feedback: str = ""
    retry_count: int = Field(default=0, ge=0)
    audit_log: Annotated[List[str], operator.add] = Field(default_factory=list)

2. Update manager_node to Calibrate & Save Weights

When the Manager decides next_agent == "ranker", it inspects the user_goal and analyzer_report to set custom multipliers and writes them into state.scoring_weights:

from langchain_core.messages import HumanMessage


def manager_node(state: TeamState) -> dict:
    structured_manager = reasoning_llm.with_structured_output(ManagerDecision).with_retry(
        stop_after_attempt=3
    )

    prompt = f"""
    Inspect the current state and route to the next specialist:
    - 'finder' if finder_data is EMPTY
    - 'analyzer' if finder_data is POPULATED and analyzer_report is EMPTY
    - 'recommender' if analyzer_report is POPULATED and recommendations is EMPTY
    - 'ranker' if recommendations is POPULATED and ranking_report is EMPTY
    - 'implementer' if ranking_report is POPULATED and implementation is EMPTY
    - 'next_steps' if implementation is POPULATED and next_steps is EMPTY
    - 'DONE' ONLY when next_steps is POPULATED

    IMPORTANT FOR 'ranker':
    When routing to 'ranker', you MUST populate `scoring_weights` (each 0.5 to 2.5) tailored to the goal:
    - Mission-critical / security / healthcare / finance -> increase `risk` (1.8-2.5) and `practicality` (1.5-2.0).
    - Rapid prototype / MVP / hackathon -> increase `feasibility` (1.8-2.5) and `benefit` (1.5-2.0), lower `complexity` penalty (0.5-0.8).
    - Long-term enterprise platform -> increase `practicality` (1.8-2.2) and `complexity` penalty (1.5-2.0).

    USER GOAL: {state.user_goal}
    ANALYZER CONSTRAINTS: {state.analyzer_report[:600] if state.analyzer_report else 'N/A'}
    STATE STATUS:
    - finder_data: {'POPULATED' if state.finder_data else 'EMPTY'}
    - analyzer_report: {'POPULATED' if state.analyzer_report else 'EMPTY'}
    - recommendations: {'POPULATED' if state.recommendations else 'EMPTY'}
    - ranking_report: {'POPULATED' if state.ranking_report else 'EMPTY'}
    - implementation: {'POPULATED' if state.implementation else 'EMPTY'}
    - next_steps: {'POPULATED' if state.next_steps else 'EMPTY'}
    """

    decision: ManagerDecision = structured_manager.invoke([HumanMessage(content=prompt)])
    print(f"\n🧠 [Manager] -> {decision.next_agent.upper()} ({decision.reasoning})")

    updates: dict = {
        "next_agent": decision.next_agent,
        "task_instructions": decision.task_instructions,
        "critic_feedback": "",
        "retry_count": 0,
        "audit_log": [f"Manager routed to {decision.next_agent}"],
    }

    if decision.next_agent == "ranker" and decision.scoring_weights:
        w = decision.scoring_weights
        print(
            f"   ⚖️ [Manager Weights] Benefit:{w.benefit}x | Feasibility:{w.feasibility}x | "
            f"Practicality:{w.practicality}x | Complexity:-{w.complexity}x | Risk:-{w.risk}x"
        )
        updates["scoring_weights"] = w
        updates["audit_log"].append(f"Manager set ranking weights: {w.weight_rationale}")

    return updates

3. Inject Dynamic Weights into RankingMatrix via ValidationInfo

Instead of hardcoding multipliers inside @computed_field, give RankedOption a net_score: float = 0.0 field and calculate it inside RankingMatrix's @model_validator(mode="after") using info.context.get("weights"):

class RankedOption(BaseModel):
    option_name: str = Field(description="Short title of the proposed option.")
    feasibility: int = Field(ge=1, le=10)
    practicality: int = Field(ge=1, le=10)
    benefit: int = Field(ge=1, le=10)
    complexity: int = Field(ge=1, le=10)
    risk: int = Field(ge=1, le=10)
    rationale: str = Field(description="Concise 1-2 sentence justification for these scores.")
    net_score: float = Field(default=0.0, description="Computed deterministically by validator.")


class RankingMatrix(BaseModel):
    evaluated_options: List[RankedOption] = Field(min_length=1)
    winning_option: str = Field(description="Exact option_name of the highest-scoring option.")
    selection_summary: str = Field(description="Why the winning option is best for the Implementer.")

    @model_validator(mode="after")
    def compute_weighted_scores(self, info: ValidationInfo) -> "RankingMatrix":
        # Pull dynamic ScoringWeights passed via model_validate(..., context={"weights": ...})
        weights: ScoringWeights = (
            info.context.get("weights") if info.context else None
        ) or ScoringWeights()

        for opt in self.evaluated_options:
            positive = (
                (opt.benefit * weights.benefit)
                + (opt.feasibility * weights.feasibility)
                + (opt.practicality * weights.practicality)
            )
            penalty = (opt.complexity * weights.complexity) + (opt.risk * weights.risk)
            opt.net_score = round(positive - penalty, 2)

        # Sort descending by dynamically weighted net_score
        self.evaluated_options.sort(key=lambda o: o.net_score, reverse=True)
        self.winning_option = self.evaluated_options[0].option_name
        return self

4. Update ranker_node to Pass state.scoring_weights

In ranker_node, fetch the raw structured output from Ollama and re-validate it with context={"weights": state.scoring_weights} so the exact Manager weights drive the math and appear in the final Markdown scorecard:

def ranker_node(state: TeamState) -> dict:
    w = state.scoring_weights
    print(f"  📊 [Ranker] Scoring options using Manager weights ({w.weight_rationale})...")

    structured_ranker = reasoning_llm.with_structured_output(RankingMatrix).with_retry(
        stop_after_attempt=3
    )

    prompt = f"""
    Score EVERY option proposed by the Recommender on an objective 1-10 raw scale:
    - Feasibility (10 = easiest to build)
    - Practicality (10 = easiest to maintain/operate)
    - Benefit (10 = highest impact toward goal)
    - Complexity (10 = most complex/over-engineered)
    - Risk (10 = highest security/failure risk)

    Note: The system will automatically apply these strategic priority multipliers to your raw scores:
    - Benefit: +{w.benefit}x | Feasibility: +{w.feasibility}x | Practicality: +{w.practicality}x
    - Complexity Penalty: -{w.complexity}x | Risk Penalty: -{w.risk}x
    - Priority Rationale: {w.weight_rationale}

    USER GOAL: {state.user_goal}
    ANALYSIS:\n{state.analyzer_report}
    PROPOSED OPTIONS:\n{state.recommendations}
    """
    if state.critic_feedback:
        prompt += f"\n\n[REVISION REQUIRED]:\n{state.critic_feedback}"

    raw_matrix: RankingMatrix = structured_ranker.invoke([HumanMessage(content=prompt)])

    # Re-validate with context so ValidationInfo injects state.scoring_weights into net_score math
    matrix = RankingMatrix.model_validate(
        raw_matrix.model_dump(),
        context={"weights": w},
    )

    table_lines = [
        f"### 🏆 Selected Strategy: **{matrix.winning_option}**",
        f"**Weighting Focus:** *{w.weight_rationale}* (`Benefit {w.benefit}x`, `Feasibility {w.feasibility}x`, `Practicality {w.practicality}x`, `Complexity -{w.complexity}x`, `Risk -{w.risk}x`)\n",
        f"{matrix.selection_summary}\n",
        "| Rank | Option | Feasibility | Practicality | Benefit | Complexity | Risk | Weighted Net Score |",
        "| :--- | :--- | :---: | :---: | :---: | :---: | :---: | :---: |",
    ]
    for idx, opt in enumerate(matrix.evaluated_options, start=1):
        table_lines.append(
            f"| #{idx} | **{opt.option_name}** | {opt.feasibility}/10 | {opt.practicality}/10 | "
            f"{opt.benefit}/10 | {opt.complexity}/10 | {opt.risk}/10 | **{opt.net_score}** |"
        )
        table_lines.append(f"> *{opt.option_name} rationale:* {opt.rationale}")

    return {
        "ranking_report": "\n".join(table_lines),
        "active_stage": "ranker",
        "audit_log": [
            f"Ranker selected '{matrix.winning_option}' (Weighted Score: {matrix.evaluated_options[0].net_score})"
        ],
    }

Why Separating Raw Scores from Weights Works So Well

• Prevents Double-Counting: The Ranker LLM is instructed to score each option objectively from 1–10 without mentally inflating numbers for priority, while Python applies the Manager's ScoringWeights cleanly afterward.

• Transparent Audit Trail: Every generated ranking_report prints the exact multipliers (Risk -2.2x, Feasibility 1.8x) right above the scorecard table so both the Execution Critic and the human user can see why Option A beat Option B.

Combine all the pieces we built (Pydantic TeamState, dynamic ScoringWeights, Ranker, Dual Critics, DuckDuckGo + Pytest tools, and LangGraph workflow) into a clean, copy-pasteable multi-file repository layout.

This complete, modular Python project combines the Hub-and-Spoke LangGraph workflow, Pydantic v2 state and validators, dynamic Manager scoring weights, the Ranker scorecard, Dual Critics, and sandboxed DuckDuckGo + Pytest tools.

Repository Layout

ollama_agent_team/
├── requirements.txt
├── schemas.py      # Pydantic state, dynamic weights, Ranker matrix, and Critic validators
├── tools.py        # Sandboxed DuckDuckGo search, ./workspace file I/O, and pytest runner
├── agents.py       # All 8 LangGraph node functions (Manager, 6 specialists, 2 Critics)
├── graph.py        # StateGraph wiring, conditional critic loops, and MemorySaver
├── main.py         # CLI entry point with streaming audit logs and final report output
└── workspace/      # Auto-created sandbox where Implementer writes & tests code

1. requirements.txt

langgraph>=0.3.0
langchain-ollama>=0.3.0
langchain-core>=0.3.0
pydantic>=2.9.0
ddgs>=7.0.0
pytest>=8.0.0

Install dependencies and pull the two local models:

pip install -r requirements.txt
ollama pull qwen3:14b
ollama pull qwen3:8b

2. schemas.py

import operator
from typing import Annotated, List, Literal, Optional
from pydantic import BaseModel, Field, ValidationInfo, field_validator, model_validator


class ScoringWeights(BaseModel):
    feasibility: float = Field(default=1.2, ge=0.5, le=2.5, description="Weight for technical/resource viability.")
    practicality: float = Field(default=1.2, ge=0.5, le=2.5, description="Weight for operational ease & maintainability.")
    benefit: float = Field(default=1.5, ge=0.5, le=2.5, description="Weight for business/user value delivered.")
    complexity: float = Field(default=0.8, ge=0.5, le=2.5, description="Penalty weight for architectural complexity.")
    risk: float = Field(default=1.0, ge=0.5, le=2.5, description="Penalty weight for security, reliability, or failure risk.")
    weight_rationale: str = Field(
        default="Standard balanced weighting.",
        description="1-sentence explanation of why these weights fit the user's goal.",
    )


class ManagerDecision(BaseModel):
    next_agent: Literal[
        "finder", "analyzer", "recommender", "ranker", "implementer", "next_steps", "DONE"
    ]
    task_instructions: str = Field(description="Specific instructions for the chosen agent.")
    reasoning: str = Field(description="Why this agent should run next.")
    scoring_weights: Optional[ScoringWeights] = Field(
        default=None,
        description="Required when next_agent == 'ranker'. Tailor weights to the user's goal and constraints.",
    )

    @model_validator(mode="after")
    def require_weights_for_ranker(self) -> "ManagerDecision":
        if self.next_agent == "ranker" and self.scoring_weights is None:
            raise ValueError(
                "Invalid ManagerDecision: You must provide 'scoring_weights' when routing to 'ranker'."
            )
        return self


class RankedOption(BaseModel):
    option_name: str = Field(description="Short title of the proposed option.")
    feasibility: int = Field(ge=1, le=10)
    practicality: int = Field(ge=1, le=10)
    benefit: int = Field(ge=1, le=10)
    complexity: int = Field(ge=1, le=10)
    risk: int = Field(ge=1, le=10)
    rationale: str = Field(description="Concise 1-2 sentence justification for these scores.")
    net_score: float = Field(default=0.0, description="Computed deterministically by validator.")


class RankingMatrix(BaseModel):
    evaluated_options: List[RankedOption] = Field(min_length=1)
    winning_option: str = Field(description="Exact option_name of the highest-scoring option.")
    selection_summary: str = Field(description="Why the winning option is best for the Implementer.")

    @model_validator(mode="after")
    def compute_weighted_scores(self, info: ValidationInfo) -> "RankingMatrix":
        weights: ScoringWeights = (
            info.context.get("weights") if info.context else None
        ) or ScoringWeights()

        for opt in self.evaluated_options:
            positive = (
                (opt.benefit * weights.benefit)
                + (opt.feasibility * weights.feasibility)
                + (opt.practicality * weights.practicality)
            )
            penalty = (opt.complexity * weights.complexity) + (opt.risk * weights.risk)
            opt.net_score = round(positive - penalty, 2)

        self.evaluated_options.sort(key=lambda o: o.net_score, reverse=True)
        self.winning_option = self.evaluated_options[0].option_name
        return self


class CriticVerdict(BaseModel):
    status: Literal["APPROVED", "REJECTED"]
    severity: Literal["LOW", "HIGH"]
    critique: List[str] = Field(
        default_factory=list,
        description="Specific flaws, hallucinations, or test failures found.",
    )
    required_fixes: str = Field(
        default="",
        description="Actionable step-by-step instructions on what must change. Required if REJECTED.",
    )

    @model_validator(mode="after")
    def enforce_rejection_actionability(self) -> "CriticVerdict":
        clean_fixes = self.required_fixes.strip()
        clean_critiques = [c.strip() for c in self.critique if c.strip()]

        if self.status == "REJECTED":
            if not clean_critiques:
                raise ValueError(
                    "Invalid CriticVerdict: When status is 'REJECTED', 'critique' must list at least one flaw."
                )
            if len(clean_fixes) < 15 or clean_fixes.lower() in {"none", "n/a", "fix it"}:
                raise ValueError(
                    "Invalid CriticVerdict: When status is 'REJECTED', 'required_fixes' must be actionable (15+ chars)."
                )

        if self.status == "APPROVED" and self.severity == "HIGH":
            raise ValueError(
                "Invalid CriticVerdict: Cannot set status='APPROVED' while severity='HIGH'."
            )

        self.required_fixes = clean_fixes
        self.critique = clean_critiques
        return self


class TeamState(BaseModel):
    user_goal: str = Field(min_length=5)
    next_agent: Literal[
        "", "finder", "analyzer", "recommender", "ranker", "implementer", "next_steps", "DONE"
    ] = ""
    task_instructions: str = ""
    finder_data: str = ""
    analyzer_report: str = ""
    recommendations: str = ""
    scoring_weights: ScoringWeights = Field(default_factory=ScoringWeights)
    ranking_report: str = ""
    implementation: str = ""
    next_steps: str = ""
    active_stage: Literal["", "analyzer", "recommender", "ranker", "implementer"] = ""
    critic_status: Literal["", "APPROVED", "REJECTED"] = ""
    critic_feedback: str = ""
    retry_count: int = Field(default=0, ge=0)
    audit_log: Annotated[List[str], operator.add] = Field(default_factory=list)

    @field_validator("user_goal")
    @classmethod
    def strip_goal(cls, v: str) -> str:
        cleaned = v.strip()
        if len(cleaned) < 5:
            raise ValueError("user_goal must contain at least 5 non-whitespace characters.")
        return cleaned

3. tools.py

import os
import subprocess
import sys
from pathlib import Path
from ddgs import DDGS
from langchain_core.tools import tool

WORKSPACE_DIR = Path("./workspace").resolve()
WORKSPACE_DIR.mkdir(parents=True, exist_ok=True)


@tool
def web_search(query: str, max_results: int = 5) -> str:
    """Search the web using DuckDuckGo for up-to-date facts, documentation, or news."""
    print(f"    🌐 [Tool: web_search] Querying: '{query}'")
    try:
        results = DDGS().text(query, max_results=max_results)
        if not results:
            return "No search results found."
        return "\n".join(
            f"- {r.get('title')}: {r.get('body')} (Source: {r.get('href')})"
            for r in results
        )
    except Exception as e:
        return f"Search error: {e}"


@tool
def write_file(filepath: str, content: str) -> str:
    """Write or overwrite a file inside the ./workspace directory."""
    print(f"    💾 [Tool: write_file] Writing: '{filepath}'")
    try:
        target_path = (WORKSPACE_DIR / filepath).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)):
            return "Security Error: Path traversal outside ./workspace is blocked."
        target_path.parent.mkdir(parents=True, exist_ok=True)
        target_path.write_text(content, encoding="utf-8")
        return f"Successfully wrote {len(content)} chars to workspace/{filepath}"
    except Exception as e:
        return f"File write error: {e}"


@tool
def read_file(filepath: str) -> str:
    """Read an existing file from the ./workspace directory."""
    print(f"    📂 [Tool: read_file] Reading: '{filepath}'")
    try:
        target_path = (WORKSPACE_DIR / filepath).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)) or not target_path.exists():
            return f"Error: File '{filepath}' does not exist in ./workspace."
        return target_path.read_text(encoding="utf-8")
    except Exception as e:
        return f"File read error: {e}"


@tool
def list_workspace_files() -> str:
    """List all files currently saved inside the ./workspace directory."""
    files = [
        str(p.relative_to(WORKSPACE_DIR))
        for p in WORKSPACE_DIR.rglob("*")
        if p.is_file() and "__pycache__" not in p.parts and ".pytest_cache" not in p.parts
    ]
    return "\n".join(sorted(files)) if files else "Workspace is currently empty."


@tool
def run_workspace_tests(target: str = "") -> str:
    """Run pytest on ./workspace or a specific test file inside it."""
    print(f"    🧪 [Tool: run_workspace_tests] Running pytest on: '{target or 'all tests'}'")
    if target:
        target_path = (WORKSPACE_DIR / target).resolve()
        if not str(target_path).startswith(str(WORKSPACE_DIR)) or not target_path.exists():
            return f"Path Error: Target '{target}' does not exist in ./workspace."
        cmd = [sys.executable, "-m", "pytest", "-v", str(target_path)]
    else:
        cmd = [sys.executable, "-m", "pytest", "-v"]

    safe_env = {
        "PATH": os.environ.get("PATH", ""),
        "PYTHONPATH": str(WORKSPACE_DIR),
        "PYTHONDONTWRITEBYTECODE": "1",
    }

    try:
        result = subprocess.run(
            cmd,
            cwd=str(WORKSPACE_DIR),
            env=safe_env,
            capture_output=True,
            text=True,
            timeout=15,
            shell=False,
        )
        output = (
            f"EXIT CODE: {result.returncode} ({'PASSED' if result.returncode == 0 else 'FAILED'})\n"
            f"--- STDOUT ---\n{result.stdout}\n"
            f"--- STDERR ---\n{result.stderr}"
        )
        return output[:2000] + "\n...[TRUNCATED]...\n" + output[-2000:] if len(output) > 4000 else output
    except subprocess.TimeoutExpired:
        return "EXECUTION ERROR: pytest timed out after 15 seconds."
    except Exception as e:
        return f"EXECUTION ERROR: {e}"

4. agents.py

Python

from langchain_ollama import ChatOllama
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import create_react_agent

from schemas import (
    TeamState,
    ManagerDecision,
    RankingMatrix,
    CriticVerdict,
)
from tools import (
    web_search,
    write_file,
    read_file,
    list_workspace_files,
    run_workspace_tests,
)

# Keep to 2 local models to avoid Ollama VRAM model-swapping overhead
reasoning_llm = ChatOllama(model="qwen3:14b", temperature=0.0)
fast_llm = ChatOllama(model="qwen3:8b", temperature=0.0)
MAX_RETRIES = 2

# Prebuilt ReAct sub-agents for multi-turn tool execution
finder_executor = create_react_agent(
    reasoning_llm,
    tools=[web_search],
    prompt=(
        "You are the Finder. Use `web_search` to gather accurate, up-to-date facts, "
        "library versions, and constraints. Cite URLs. Do NOT propose solutions."
    ),
)

implementer_executor = create_react_agent(
    reasoning_llm,
    tools=[write_file, read_file, list_workspace_files, run_workspace_tests],
    prompt=(
        "You are the Implementer. Execute ONLY the winning strategy from the Ranking Report. "
        "Write complete, production-ready code into ./workspace using `write_file`, write unit "
        "tests in `tests/`, and run `run_workspace_tests` until EXIT CODE is 0. "
        "Return a concise summary of created files and test results."
    ),
)

qa_auditor_executor = create_react_agent(
    reasoning_llm,
    tools=[list_workspace_files, read_file, run_workspace_tests],
    prompt=(
        "You are a Senior QA & Security Auditor. Inspect ./workspace files using "
        "`list_workspace_files` and `read_file`, and run `run_workspace_tests` to "
        "independently verify that tests exist, cover edge cases, and pass."
    ),
)


def manager_node(state: TeamState) -> dict:
    structured_manager = reasoning_llm.with_structured_output(ManagerDecision).with_retry(
        stop_after_attempt=3
    )

    prompt = f"""
    Inspect the current state and route to the next specialist:
    - 'finder' if finder_data is EMPTY
    - 'analyzer' if finder_data is POPULATED and analyzer_report is EMPTY
    - 'recommender' if analyzer_report is POPULATED and recommendations is EMPTY
    - 'ranker' if recommendations is POPULATED and ranking_report is EMPTY
    - 'implementer' if ranking_report is POPULATED and implementation is EMPTY
    - 'next_steps' if implementation is POPULATED and next_steps is EMPTY
    - 'DONE' ONLY when next_steps is POPULATED

    IMPORTANT FOR 'ranker':
    When routing to 'ranker', you MUST populate `scoring_weights` (each 0.5 to 2.5) tailored to the goal:
    - Mission-critical / security / finance -> increase `risk` (1.8-2.5) and `practicality` (1.5-2.0).
    - Rapid prototype / MVP -> increase `feasibility` (1.8-2.5) and `benefit` (1.5-2.0), lower `complexity` penalty (0.5-0.8).
    - Long-term production system -> increase `practicality` (1.8-2.2) and `complexity` penalty (1.5-2.0).

    USER GOAL: {state.user_goal}
    ANALYZER CONSTRAINTS: {state.analyzer_report[:600] if state.analyzer_report else 'N/A'}
    STATE STATUS:
    - finder_data: {'POPULATED' if state.finder_data else 'EMPTY'}
    - analyzer_report: {'POPULATED' if state.analyzer_report else 'EMPTY'}
    - recommendations: {'POPULATED' if state.recommendations else 'EMPTY'}
    - ranking_report: {'POPULATED' if state.ranking_report else 'EMPTY'}
    - implementation: {'POPULATED' if state.implementation else 'EMPTY'}
    - next_steps: {'POPULATED' if state.next_steps else 'EMPTY'}
    """

    decision: ManagerDecision = structured_manager.invoke([HumanMessage(content=prompt)])
    print(f"\n🧠 [Manager] -> {decision.next_agent.upper()} ({decision.reasoning})")

    updates: dict = {
        "next_agent": decision.next_agent,
        "task_instructions": decision.task_instructions,
        "critic_feedback": "",
        "retry_count": 0,
        "audit_log": [f"Manager routed to {decision.next_agent}"],
    }

    if decision.next_agent == "ranker" and decision.scoring_weights:
        w = decision.scoring_weights
        print(
            f"   ⚖️ [Manager Weights] Benefit:{w.benefit}x | Feasibility:{w.feasibility}x | "
            f"Practicality:{w.practicality}x | Complexity:-{w.complexity}x | Risk:-{w.risk}x"
        )
        updates["scoring_weights"] = w
        updates["audit_log"].append(f"Manager set ranking weights: {w.weight_rationale}")

    return updates


def finder_node(state: TeamState) -> dict:
    print("  🔎 [Finder] Gathering facts and documentation...")
    msg = f"Goal: {state.user_goal}\nInstructions: {state.task_instructions}"
    res = finder_executor.invoke({"messages": [HumanMessage(content=msg)]})
    return {
        "finder_data": res["messages"][-1].content,
        "audit_log": ["Finder gathered research context"],
    }


def analyzer_node(state: TeamState) -> dict:
    print("  📐 [Analyzer] Synthesizing patterns and constraints...")
    sys_msg = SystemMessage(
        content="You are the Analyzer. Synthesize Finder data into patterns, constraints, and root causes. Do NOT propose solutions."
    )
    user_msg = (
        f"Goal: {state.user_goal}\n"
        f"Instructions: {state.task_instructions}\n"
        f"Finder Data:\n{state.finder_data}"
    )
    if state.critic_feedback:
        user_msg += f"\n\n[REVISION REQUIRED]:\n{state.critic_feedback}"

    res = reasoning_llm.invoke([sys_msg, HumanMessage(content=user_msg)])
    return {
        "analyzer_report": res.content,
        "active_stage": "analyzer",
        "audit_log": ["Analyzer completed diagnostic report"],
    }


def recommender_node(state: TeamState) -> dict:
    print("  💡 [Recommender] Generating strategic options...")
    sys_msg = SystemMessage(
        content="You are the Recommender. Propose 2-3 distinct architectural options with pros, cons, and trade-offs."
    )
    user_msg = (
        f"Goal: {state.user_goal}\n"
        f"Instructions: {state.task_instructions}\n"
        f"Analysis:\n{state.analyzer_report}"
    )
    if state.critic_feedback:
        user_msg += f"\n\n[REVISION REQUIRED]:\n{state.critic_feedback}"

    res = reasoning_llm.invoke([sys_msg, HumanMessage(content=user_msg)])
    return {
        "recommendations": res.content,
        "active_stage": "recommender",
        "audit_log": ["Recommender proposed strategic options"],
    }


def ranker_node(state: TeamState) -> dict:
    w = state.scoring_weights
    print(f"  📊 [Ranker] Scoring options ({w.weight_rationale})...")
    structured_ranker = reasoning_llm.with_structured_output(RankingMatrix).with_retry(
        stop_after_attempt=3
    )

    prompt = f"""
    Score EVERY option proposed by the Recommender on an objective 1-10 raw scale:
    - Feasibility (10 = easiest to build)
    - Practicality (10 = easiest to maintain/operate)
    - Benefit (10 = highest impact toward goal)
    - Complexity (10 = most complex/over-engineered)
    - Risk (10 = highest security/failure risk)

    Priority Multipliers Applied Automatically:
    - Benefit: +{w.benefit}x | Feasibility: +{w.feasibility}x | Practicality: +{w.practicality}x
    - Complexity Penalty: -{w.complexity}x | Risk Penalty: -{w.risk}x
    - Rationale: {w.weight_rationale}

    USER GOAL: {state.user_goal}
    ANALYSIS:\n{state.analyzer_report}
    PROPOSED OPTIONS:\n{state.recommendations}
    """
    if state.critic_feedback:
        prompt += f"\n\n[REVISION REQUIRED]:\n{state.critic_feedback}"

    raw_matrix: RankingMatrix = structured_ranker.invoke([HumanMessage(content=prompt)])
    matrix = RankingMatrix.model_validate(
        raw_matrix.model_dump(),
        context={"weights": w},
    )

    table_lines = [
        f"### 🏆 Selected Strategy: **{matrix.winning_option}**",
        f"**Weighting Focus:** *{w.weight_rationale}* (`Benefit {w.benefit}x`, `Feasibility {w.feasibility}x`, `Practicality {w.practicality}x`, `Complexity -{w.complexity}x`, `Risk -{w.risk}x`)\n",
        f"{matrix.selection_summary}\n",
        "| Rank | Option | Feasibility | Practicality | Benefit | Complexity | Risk | Weighted Net Score |",
        "| :--- | :--- | :---: | :---: | :---: | :---: | :---: | :---: |",
    ]
    for idx, opt in enumerate(matrix.evaluated_options, start=1):
        table_lines.append(
            f"| #{idx} | **{opt.option_name}** | {opt.feasibility}/10 | {opt.practicality}/10 | "
            f"{opt.benefit}/10 | {opt.complexity}/10 | {opt.risk}/10 | **{opt.net_score}** |"
        )
        table_lines.append(f"> *{opt.option_name} rationale:* {opt.rationale}")

    return {
        "ranking_report": "\n".join(table_lines),
        "active_stage": "ranker",
        "audit_log": [
            f"Ranker selected '{matrix.winning_option}' (Score: {matrix.evaluated_options[0].net_score})"
        ],
    }


def implementer_node(state: TeamState) -> dict:
    print("  🛠️ [Implementer] Building code and running pytest in ./workspace...")
    user_msg = (
        f"Goal: {state.user_goal}\n"
        f"Instructions: {state.task_instructions}\n"
        f"Winning Strategy Scorecard:\n{state.ranking_report}\n\n"
        f"Full Option Details:\n{state.recommendations}"
    )
    if state.critic_feedback:
        user_msg += f"\n\n[REVISION REQUIRED - FIX THESE ISSUES]:\n{state.critic_feedback}"

    res = implementer_executor.invoke(
        {"messages": [HumanMessage(content=user_msg)]},
        config={"recursion_limit": 25},
    )
    return {
        "implementation": res["messages"][-1].content,
        "active_stage": "implementer",
        "audit_log": ["Implementer built and tested code in ./workspace"],
    }


def logic_critic_node(state: TeamState) -> dict:
    print("  🔍 [Logic & Fact Critic] Auditing analysis...")
    structured_critic = reasoning_llm.with_structured_output(CriticVerdict).with_retry(
        stop_after_attempt=3
    )
    prompt = (
        "Evaluate the Analyzer report for hallucinations, unsupported leaps, or contradictions. "
        "Only approve if zero HIGH severity flaws exist.\n\n"
        f"Goal: {state.user_goal}\nFinder Data:\n{state.finder_data}\n"
        f"Analyzer Report:\n{state.analyzer_report}"
    )
    verdict = structured_critic.invoke([HumanMessage(content=prompt)])
    return _format_critic_update("Logic & Fact", verdict, state)


def execution_critic_node(state: TeamState) -> dict:
    stage = state.active_stage
    print(f"  🔍 [Execution & Risk Critic] Auditing {stage}...")
    structured_critic = reasoning_llm.with_structured_output(CriticVerdict).with_retry(
        stop_after_attempt=3
    )

    if stage in {"recommender", "ranker"}:
        artifact = state.recommendations if stage == "recommender" else state.ranking_report
        prompt = (
            f"Evaluate the {stage} output for feasibility, edge cases, and constraint alignment.\n\n"
            f"Goal: {state.user_goal}\nOutput to Review:\n{artifact}"
        )
        verdict = structured_critic.invoke([HumanMessage(content=prompt)])
        return _format_critic_update("Execution & Risk", verdict, state)

    # Stage == "implementer": Run independent QA audit + pytest first
    audit_res = qa_auditor_executor.invoke(
        {
            "messages": [
                HumanMessage(
                    content=(
                        f"Goal: {state.user_goal}\nImplementer Summary:\n{state.implementation}\n"
                        "Inspect ./workspace files and run pytest now."
                    )
                )
            ]
        },
        config={"recursion_limit": 15},
    )
    audit_report = audit_res["messages"][-1].content

    verdict_prompt = (
        "Based on the Implementer summary and the independent QA/pytest audit below, issue a verdict. "
        "If pytest failed or tests are missing/trivial, set status='REJECTED' and severity='HIGH'.\n\n"
        f"Goal: {state.user_goal}\nImplementer Summary:\n{state.implementation}\n"
        f"Independent QA & Pytest Audit:\n{audit_report}"
    )
    verdict = structured_critic.invoke([HumanMessage(content=verdict_prompt)])
    return _format_critic_update("Execution & Risk", verdict, state)


def _format_critic_update(critic_name: str, verdict: CriticVerdict, state: TeamState) -> dict:
    if verdict.status == "APPROVED":
        print(f"  ✅ [{critic_name} Critic] APPROVED.")
        return {
            "critic_status": "APPROVED",
            "critic_feedback": "",
            "audit_log": [f"{critic_name} Critic APPROVED {state.active_stage}"],
        }

    new_retries = state.retry_count + 1
    print(f"  ❌ [{critic_name} Critic] REJECTED (Attempt {new_retries}/{MAX_RETRIES}): {verdict.critique}")
    if new_retries > MAX_RETRIES:
        print("  ⚠️ Max retries reached. Force-approving to prevent infinite loop.")
        return {
            "critic_status": "APPROVED",
            "critic_feedback": "",
            "audit_log": [f"{critic_name} Critic max retries reached on {state.active_stage}"],
        }

    return {
        "critic_status": "REJECTED",
        "critic_feedback": f"Critique: {verdict.critique}\nRequired Fixes: {verdict.required_fixes}",
        "retry_count": new_retries,
        "audit_log": [f"{critic_name} Critic REJECTED {state.active_stage} (Attempt {new_retries})"],
    }


def next_steps_node(state: TeamState) -> dict:
    print("  📋 [Next Steps] Packaging final deliverable and user checklist...")
    sys_msg = SystemMessage(
        content="You are the Next Steps Coordinator. Provide: (1) Executive Summary, (2) Chronological Action Checklist, (3) 2-3 Verification KPIs."
    )
    user_msg = (
        f"Goal: {state.user_goal}\n"
        f"Winning Strategy:\n{state.ranking_report}\n"
        f"Implementation:\n{state.implementation}"
    )
    res = fast_llm.invoke([sys_msg, HumanMessage(content=user_msg)])
    return {
        "next_steps": res.content,
        "audit_log": ["Next Steps packaged final handoff"],
    }

5. graph.py

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

from schemas import TeamState
from agents import (
    manager_node,
    finder_node,
    analyzer_node,
    recommender_node,
    ranker_node,
    implementer_node,
    logic_critic_node,
    execution_critic_node,
    next_steps_node,
)


def route_from_manager(state: TeamState) -> str:
    return END if state.next_agent == "DONE" else state.next_agent


def route_from_critic(state: TeamState) -> str:
    return "manager" if state.critic_status == "APPROVED" else state.active_stage


def build_graph(interrupt_before_implementer: bool = False):
    workflow = StateGraph(TeamState)

    # Register all 8 agent nodes
    workflow.add_node("manager", manager_node)
    workflow.add_node("finder", finder_node)
    workflow.add_node("analyzer", analyzer_node)
    workflow.add_node("recommender", recommender_node)
    workflow.add_node("ranker", ranker_node)
    workflow.add_node("implementer", implementer_node)
    workflow.add_node("logic_critic", logic_critic_node)
    workflow.add_node("execution_critic", execution_critic_node)
    workflow.add_node("next_steps", next_steps_node)

    # Entry point
    workflow.add_edge(START, "manager")

    # Manager hub routing
    workflow.add_conditional_edges(
        "manager",
        route_from_manager,
        {
            "finder": "finder",
            "analyzer": "analyzer",
            "recommender": "recommender",
            "ranker": "ranker",
            "implementer": "implementer",
            "next_steps": "next_steps",
            END: END,
        },
    )

    # Forward pipeline edges
    workflow.add_edge("finder", "manager")
    workflow.add_edge("analyzer", "logic_critic")
    workflow.add_edge("recommender", "execution_critic")
    workflow.add_edge("ranker", "execution_critic")
    workflow.add_edge("implementer", "execution_critic")
    workflow.add_edge("next_steps", "manager")

    # Critic quality-gate loops
    workflow.add_conditional_edges(
        "logic_critic",
        route_from_critic,
        {"manager": "manager", "analyzer": "analyzer"},
    )
    workflow.add_conditional_edges(
        "execution_critic",
        route_from_critic,
        {
            "manager": "manager",
            "recommender": "recommender",
            "ranker": "ranker",
            "implementer": "implementer",
        },
    )

    interrupts = ["implementer"] if interrupt_before_implementer else []
    return workflow.compile(
        checkpointer=MemorySaver(),
        interrupt_before=interrupts,
    )

6. main.py

import uuid
from graph import build_graph
from schemas import TeamState


def run_project(goal: str):
    app = build_graph(interrupt_before_implementer=False)
    thread_id = f"run-{uuid.uuid4().hex[:8]}"
    config = {"configurable": {"thread_id": thread_id}}

    print(f"🚀 Starting AI Agent Team (Thread: {thread_id})")
    print(f"🎯 Goal: {goal}\n" + "=" * 70)

    # Invoke with minimal dict; Pydantic populates defaults & validates input
    final_raw = app.invoke({"user_goal": goal}, config=config)
    final_state = TeamState.model_validate(final_raw)

    print("\n" + "=" * 70)
    print("📊 RANKING SCORECARD:\n")
    print(final_state.ranking_report)

    print("\n" + "=" * 70)
    print("🛠️ IMPLEMENTATION SUMMARY (Files saved in ./workspace):\n")
    print(final_state.implementation)

    print("\n" + "=" * 70)
    print("📋 NEXT STEPS & KPIS:\n")
    print(final_state.next_steps)

    print("\n" + "=" * 70)
    print("🧾 AUDIT TRAIL:")
    for step in final_state.audit_log:
        print(f"  - {step}")


if __name__ == "__main__":
    user_goal = (
        "Build a thread-safe sliding-window rate limiter in Python with configurable "
        "per-client limits and a complete pytest unit test suite."
    )
    run_project(user_goal)

Running & Customizing

• Start a run: Execute python main.py. All generated source files and tests will appear inside ./workspace, verified by pytest before the Execution Critic approves them.

• Enable Human-in-the-Loop Approval: Change interrupt_before_implementer=True in main.py to pause execution right after the Ranker and Execution Critic finish, letting you inspect app.get_state(config).values["ranking_report"] before calling app.invoke(None, config=config) to resume code generation.

Update main.py to use app.stream(..., stream_mode='updates') so each agent's output prints incrementally as nodes complete.

In LangGraph, app.stream(..., stream_mode="updates") yields a dictionary {node_name: state_delta} immediately as each node finishes executing, rather than blocking until the entire graph reaches END.

Because you configured MemorySaver() in graph.py, you can stream every incremental node delta to the console in real time and then call app.get_state(config).values at the end to hydrate the complete TeamState object—including handling interrupt_before_implementer=True cleanly if enabled.

Updated main.py

import time
import uuid
from graph import build_graph
from schemas import TeamState

# Map node names to the state keys they populate so we can stream clean previews
NODE_ARTIFACT_KEYS = {
    "finder": ("finder_data", "🔎 FINDER RESEARCH"),
    "analyzer": ("analyzer_report", "📐 ANALYZER DIAGNOSTICS"),
    "recommender": ("recommendations", "💡 STRATEGIC OPTIONS"),
    "ranker": ("ranking_report", "📊 WEIGHTED RANKING SCORECARD"),
    "implementer": ("implementation", "🛠️ IMPLEMENTATION SUMMARY"),
    "next_steps": ("next_steps", "📋 NEXT STEPS & KPIS"),
}


def run_project(goal: str, human_approval: bool = False):
    app = build_graph(interrupt_before_implementer=human_approval)
    thread_id = f"run-{uuid.uuid4().hex[:8]}"
    config = {"configurable": {"thread_id": thread_id}}

    print(f"🚀 Starting AI Agent Team (Thread: {thread_id})")
    print(f"🎯 Goal: {goal}\n" + "=" * 70)

    start_time = time.perf_counter()
    stream_input: dict | None = {"user_goal": goal}

    while True:
        # stream_mode="updates" yields {node_name: partial_state_dict} as each node finishes
        for chunk in app.stream(stream_input, config=config, stream_mode="updates"):
            for node_name, delta in chunk.items():
                elapsed = time.perf_counter() - start_time
                print(f"\n⏱️  [{elapsed:6.1f}s] Completed Node: {node_name.upper()}")

                # Print any new audit log entries emitted by this node
                for entry in delta.get("audit_log", []):
                    print(f"    • {entry}")

                # Stream the major artifact produced by specialist nodes as soon as they finish
                if node_name in NODE_ARTIFACT_KEYS:
                    state_key, banner = NODE_ARTIFACT_KEYS[node_name]
                    content = delta.get(state_key, "")
                    if content:
                        print("\n" + "-" * 70)
                        print(f"{banner}:")
                        print("-" * 70)
                        print(content)
                        print("-" * 70)

                # Print Critic feedback immediately if a quality gate rejects an output
                if node_name in {"logic_critic", "execution_critic"}:
                    status = delta.get("critic_status")
                    if status == "REJECTED":
                        print(f"\n⚠️  CRITIC FEEDBACK SENT BACK TO AGENT:\n{delta.get('critic_feedback')}")

        # Check if the graph paused at an interrupt (e.g., interrupt_before=['implementer'])
        snapshot = app.get_state(config)
        if snapshot.next:
            print(f"\n⏸️  PAUSED BEFORE: {snapshot.next}")
            choice = input("Press [Enter] to approve and continue, or type 'q' to abort: ").strip().lower()
            if choice == "q":
                print("🛑 Run aborted by user.")
                return
            # Passing None resumes execution from the saved checkpoint
            stream_input = None
        else:
            break

    # Retrieve the complete accumulated state from the MemorySaver checkpoint
    final_state = TeamState.model_validate(app.get_state(config).values)
    total_time = time.perf_counter() - start_time

    print("\n" + "=" * 70)
    print(f"🏁 WORKFLOW COMPLETE in {total_time:.1f}s | Full Audit Trail:")
    for idx, step in enumerate(final_state.audit_log, start=1):
        print(f"  {idx}. {step}")


if __name__ == "__main__":
    user_goal = (
        "Build a thread-safe sliding-window rate limiter in Python with configurable "
        "per-client limits and a complete pytest unit test suite."
    )
    # Set human_approval=True to pause and review the Ranker's scorecard before coding begins
    run_project(user_goal, human_approval=False)

How stream_mode="updates" Changes Execution Flow

• Immediate Visibility: Instead of waiting 2–4 minutes in silence while Finder, Analyzer, Recommender, Ranker, Implementer, and Critics run sequentially, each node's formatted report prints the exact second that node completes.

• Seamless Interrupt Resumption: When human_approval=True is set, the for chunk in app.stream(...) loop exits cleanly at the implementer breakpoint. Checking app.get_state(config).next lets you prompt the user in the terminal and resume the stream from the exact checkpoint by passing None into the next app.stream(None, ...) call.

• No Double-Execution on Final State: Because MemorySaver stores the cumulative state across every streamed chunk, calling app.get_state(config).values at the end retrieves the full TeamState without re-running the graph.

From <https://gemini.google.com/app/56bdc3b67486432e>  Google Gemini (3.8 Flash)