Self-Healing Multi-Agent Team in LangGraph: Architect, Tester, Debugger

AI now sits in nearly every layer of software work, from generating code to analyzing data to multi-step reasoning. Yet anyone who asks a large language model for something complex runs into the same snag. A single-shot answer can look right on the surface and then fall apart once it meets the requirements and a few edge cases. The usual response is a manual loop: copy the error log, paste it into the chat, ask for a fix, and repeat until things finally work.

The opposite approach has its own problem. Handing one monolithic agent dozens of tools and a long list of instructions tends to raise the error rate, because its context gets polluted and tool signatures start to blur together. Output quality drops since the agent can't follow all those instructions at once.

The design below answers both problems with a modular, hierarchical multi-agent state machine written in Python with LangGraph. The running example is a self-healing software engineering team made of five roles: an Architect, a Developer, a subprocess-based Pytest sandbox, a Debugger and a Reviewer. The pattern itself carries over to many other AI engineering workflows.

Splitting a big job among specialized sub-agents, giving each the right amount of reasoning, running the results in a controlled sandbox and adding a steady feedback loop lets the team test its own work, spot problems and repair them before handing over a verified result. First, a look at the architecture.

Two principles shape the system.

The first is heterogeneous model tiering with adaptive thinking levels. Roles differ in difficulty, so they shouldn't all get the same model or the same compute budget:

• The Architect runs on gemini-pro-latest with a high thinking level. It does the most reasoning-heavy work, such as setting problem boundaries, anticipating race conditions and designing strict pytest suites.
• The Developer runs on gemini-3.7-flash with medium thinking. It's a balanced setup for fast code synthesis that follows the Architect's explicit specification.
• The Test Runner is a deterministic Python subprocess that runs pytest locally in roughly 15 milliseconds and spends no LLM API tokens at all.
• The Debugger uses gemini-3.7-flash with high thinking, since tracing raw stack traces back to the buggy line calls for causal root-cause analysis.
• The Reviewer uses gemini-flash-lite-latest with low thinking. Checking style and writing the closing summary for the user needs only light reasoning.

The second principle is strict tool scoping, which is least privilege applied to agents. No sub-agent gets unrestricted access to the terminal and file system. Each receives only the inputs and tools its role requires. The Developer can't run arbitrary shell commands, and the Test Runner just executes the tests and reports what actually happened, without asking an LLM to interpret or invent the outcome. Narrow scope keeps each agent focused and cuts down hallucination.

The LangGraph concepts involved

LangGraph works like a shared whiteboard. Each team member, meaning each sub-agent, does its assigned job and writes the outcome back to the same shared space, so everyone sees the context the team has built so far. LangGraph is the orchestrator: it decides which agent runs next and passes the relevant state along.

Four ideas matter here:

1. State, a TypedDict, is the whiteboard. All agents read from and write to one Python dictionary called AgentTeamState, which holds the user prompt, the specification, the unit tests, the current code draft, the test results and the count of debugging rounds.
2. Nodes are the team members. A node is just a Python function that receives the current state, does its job (calling an LLM, running tests) and returns updates that get merged back into the shared state.
3. Edges carry the workflow logic. A direct edge is a fixed transition, such as always moving from the Architect to the Developer. A conditional edge decides from the current state, for example routing to the Reviewer when all tests pass and to the Debugger when any fail.
4. The compiled graph is what you get after registering nodes and edges with StateGraph and compiling. It can be run, and its events can be streamed as the workflow progresses.

Setting up the project

Start with a fresh directory and a virtual environment. These terminal commands create both and activate the environment:

mkdir langgraph-self-healing-team
cd langgraph-self-healing-team
python3 -m venv .venv
source .venv/bin/activate

A requirements.txt file then lists the dependencies:

langgraph>=0.2.20
langchain-google-genai>=2.0.0
langchain-openai>=1.6.0
langchain-core>=0.3.0
pydantic>=2.7.0
pytest>=8.0.0
rich>=13.7.0
python-dotenv>=1.0.1

Installing them into the virtual environment is one command:

pip install -r requirements.txt

Each package has a job:

• langgraph orchestrates the multi-agent workflow, handling state, handoffs between nodes and cyclic feedback loops.
• langchain-google-genai connects to Google Gemini models and exposes their thinking levels.
• langchain-openai is optional. It adds OpenRouter or OpenAI support, so models like GPT 5.6 or Deepseek v4 Pro can be swapped in if relying only on Gemini isn't desirable.
• pytest is the test runner inside the local subprocess sandbox. It runs the generated suite and returns real results.
• rich formats terminal output with colors, tables and syntax highlighting, which makes the agent workflow easier to follow and debug.
• python-dotenv loads API keys and model settings from the .env file automatically.

The .env file in the project root holds credentials and model configuration:

# API Keys
GEMINI_API_KEY="your_api_key_here"
# OPENROUTER_API_KEY=""
# OPENAI_API_KEY=""

# Model Tier Selection & Adaptive Thinking Levels (low, medium, high)
ARCHITECT_MODEL="gemini-pro-latest"
ARCHITECT_THINKING_LEVEL="high"

CODER_MODEL="gemini-3.7-flash"
CODER_THINKING_LEVEL="medium"

DEBUGGER_MODEL="gemini-3.7-flash"
DEBUGGER_THINKING_LEVEL="high"

REVIEWER_MODEL="gemini-flash-lite-latest"
REVIEWER_THINKING_LEVEL="low"

# Maximum Self-Healing Retry Loops
MAX_DEBUG_ITERATIONS=3

With the environment ready, the application can be built piece by piece.

Step 1: the shared state in src/state.py

Before anything else, the schema for data passed between agents needs defining. Create a src directory with a state.py file inside it.

The first definition is a TestResult model, written with Pydantic. It represents the structured outcome of a pytest run in the sandbox:

from typing import TypedDict, List, Dict, Any, Optional
from pydantic import BaseModel, Field

class TestResult(BaseModel):
    passed: bool = Field(description="Whether all tests in the test suite passed")
    exit_code: int = Field(description="Exit code of the pytest execution process")
    output: str = Field(description="Captured stdout and stderr from pytest")
    passed_count: int = Field(default=0, description="Number of passed tests")
    failed_count: int = Field(default=0, description="Number of failed tests")
    error_summary: Optional[str] = Field(default=None, description="Concise error traceback summary")

TestResult records whether the suite passed, the process exit code, how many tests passed and failed, and the raw test output. It also carries an error_summary field with just the relevant part of the traceback, which the Debugger uses to find and fix the failure.

Next comes the main AgentTeamState as a TypedDict. This is the whiteboard that moves through the graph: every node reads what it needs, does its work and updates the fields that follow-up nodes depend on.

class AgentTeamState(TypedDict):
    """
    The shared whiteboard state passed between all nodes in the LangGraph team.
    """
    user_request: str
    specs: str
    tests: str
    code: str
    test_results: Dict[str, Any]
    iteration: int
    max_iterations: int
    debug_history: List[str]
    review_summary: str
    status: str

Each field has a role in the workflow:

• user_request holds the original problem from the user, so every agent can see the initial intent throughout the run.
• specs and tests come from the Architect. The Developer implements against the specification, and the Test Runner executes the generated tests against the code.
• code is the current Python implementation. The Developer fills it in first, and the Debugger may replace it after a failed run.
• test_results stores the structured result from the sandbox, which drives the choice between the Reviewer and the Debugger.
• iteration and max_iterations track the self-healing loop. They keep the Debugger from retrying forever and give a safety cap, such as three attempts.
• debug_history keeps the root cause and the fix from each debugging round, a record of how the code changed.
• review_summary receives the Reviewer's final report once the code has passed validation.
• status shows where the workflow currently is: generating code, running tests, debugging or ready for review.

Step 2: model tiers and thinking levels in src/config.py

A unified LLM factory assigns the right model and reasoning level (low, medium or high) to every sub-agent role. The file src/config.py starts with a default configuration dictionary:

import os
from typing import Dict, Any, Optional
from dotenv import load_dotenv
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_openai import ChatOpenAI

load_dotenv()

# Role Configurations & Adaptive Thinking Levels
DEFAULT_CONFIG: Dict[str, Dict[str, Any]] = {
    "architect": {
        "model": os.getenv("ARCHITECT_MODEL", "gemini-pro-latest"),
        "thinking_level": os.getenv("ARCHITECT_THINKING_LEVEL", "high"),
        "temperature": 0.2,
    },
    "coder": {
        "model": os.getenv("CODER_MODEL", "gemini-3.7-flash"),
        "thinking_level": os.getenv("CODER_THINKING_LEVEL", "medium"),
        "temperature": 0.2,
    },
    "debugger": {
        "model": os.getenv("DEBUGGER_MODEL", "gemini-3.7-flash"),
        "thinking_level": os.getenv("DEBUGGER_THINKING_LEVEL", "high"),
        "temperature": 0.1,
    },
    "reviewer": {
        "model": os.getenv("REVIEWER_MODEL", "gemini-flash-lite-latest"),
        "thinking_level": os.getenv("REVIEWER_THINKING_LEVEL", "low"),
        "temperature": 0.2,
    },
}

MAX_DEBUG_ITERATIONS = int(os.getenv("MAX_DEBUG_ITERATIONS", "3"))

Each role gets a model and a thinking level that match the difficulty of its task, which balances reasoning strength, speed and cost without hurting overall quality:

• architect and debugger are set to high, because system design and stack-trace diagnosis both need deep multi-step deduction.
• coder is set to medium. It only has to follow the plan the Architect produced, so balanced, quick and grounded synthesis is enough.
• reviewer is set to low to save tokens, since all it does is write a summary after looking over the results.

The same file also holds the get_llm() factory:

def get_llm(
    role: str,
    model_override: Optional[str] = None,
    thinking_level_override: Optional[str] = None,
) -> BaseChatModel:
    """
    Unified LLM Factory. Returns an initialized LangChain ChatModel configured
    specifically for the sub-agent role with adaptive thinking levels.
    """
    role_conf = DEFAULT_CONFIG.get(role, DEFAULT_CONFIG["coder"])
    model_name = model_override or role_conf["model"]
    thinking_level = thinking_level_override or role_conf["thinking_level"]
    temp = role_conf["temperature"]

    openrouter_key = os.getenv("OPENROUTER_API_KEY")
    openai_key = os.getenv("OPENAI_API_KEY")
    gemini_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")

    # 1. OpenRouter Provider (Handles open models like deepseek/deepseek-r1:free)
    if openrouter_key and ("/" in model_name or not gemini_key):
        return ChatOpenAI(
            model=model_name,
            api_key=openrouter_key,
            base_url="https://openrouter.ai/api/v1",
            temperature=temp,
        )

    # 2. Google Gemini Provider
    if "gemini" in model_name.lower() or gemini_key:
        kwargs: Dict[str, Any] = {
            "model": model_name,
            "temperature": temp,
        }
        if gemini_key:
            kwargs["google_api_key"] = gemini_key

        if thinking_level and thinking_level.lower() in ["low", "medium", "high"]:
            kwargs["thinking_level"] = thinking_level.lower()

        return ChatGoogleGenerativeAI(**kwargs)

    # 3. Direct OpenAI Provider Fallback
    if openai_key:
        return ChatOpenAI(
            model=model_name,
            api_key=openai_key,
            temperature=temp,
        )

    return ChatGoogleGenerativeAI(model=model_name, temperature=temp)

Two things make this function useful. It passes the thinking_level (low, medium, high) straight to ChatGoogleGenerativeAI, so reasoning effort can be tuned per role. And every client it returns implements LangChain's BaseChatModel, so graph nodes call .invoke() the same way no matter which provider sits behind a role.

Step 3: a deterministic pytest sandbox in src/sandbox.py

A frequent mistake in agent design is to ask an LLM to verify, in its head, the code it just wrote. A model can confidently predict that the code passes while bugs sit inside it, because models regularly hallucinate that their own output is correct. Even when one does flag some bugs, there's no guarantee it caught them all.

A more dependable route is letting the code prove itself. Instead of trusting the model's judgment, a local execution node writes the generated code and tests into a temporary directory and runs pytest in an isolated Python subprocess. The result is an objective pass or fail that can feed straight back into the workflow.

Create sandbox.py in the src folder. It begins with a helper, _extract_error_summary(), which pulls the most relevant failure details from the raw pytest output so the Debugger sees the useful part of the traceback and not the whole dump:

import os
import re
import sys
import tempfile
import subprocess
from typing import Dict, Any

def _extract_error_summary(pytest_output: str) -> str:
    """
    Extracts the most relevant failure section (FAILURES / ERRORS traceback)
    from the raw pytest output to provide dense, actionable signal to the Debugger.
    """
    if "=== FAILURES ===" in pytest_output:
        failures_part = pytest_output.split("=== FAILURES ===")[-1]
        if "=== short test summary info ===" in failures_part:
            failures_part = failures_part.split("=== short test summary info ===")[0]
        return failures_part.strip()

    if "=== ERRORS ===" in pytest_output:
        return pytest_output.split("=== ERRORS ===")[-1].strip()

    # Fallback to the last 25 lines
    lines = pytest_output.splitlines()
    return "\n".join(lines[-25:])

The helper strips terminal color codes and boilerplate test headers, leaving the exact assertion failure (assert result == expected) plus the stack trace. The Debugger therefore gets a short, high-signal description of what went wrong, with no wasted tokens.

The main runner, run_tests_in_sandbox(), goes in the same file:

def run_tests_in_sandbox(code: str, tests: str, timeout_seconds: int = 15) -> Dict[str, Any]:
    """
    Executes unit tests against the generated code in an isolated temporary directory.
    Runs locally in a Python subprocess with zero LLM API calls ($0.00 / 0 tokens).
    """
    with tempfile.TemporaryDirectory() as temp_dir:
        code_file = os.path.join(temp_dir, "solution.py")
        test_file = os.path.join(temp_dir, "test_solution.py")

        # Write solution code and test files into the temporary directory
        with open(code_file, "w", encoding="utf-8") as f:
            f.write(code)

        with open(test_file, "w", encoding="utf-8") as f:
            f.write(tests)

        # Build execution command using the active Python interpreter
        python_exec = sys.executable
        cmd = [python_exec, "-m", "pytest", "-v", "test_solution.py"]

        try:
            result = subprocess.run(
                cmd,
                cwd=temp_dir,
                capture_output=True,
                text=True,
                timeout=timeout_seconds,
            )
            output = (result.stdout + "\n" + result.stderr).strip()
            exit_code = result.returncode

        except subprocess.TimeoutExpired as e:
            output = f"Execution Timed Out after {timeout_seconds} seconds! Possible infinite loop.\n" + (e.stdout or "")
            exit_code = -1

        passed = (exit_code == 0)

        passed_count = 0
        failed_count = 0

        passed_match = re.search(r"(\d+)\s+passed", output)
        if passed_match:
            passed_count = int(passed_match.group(1))

        failed_match = re.search(r"(\d+)\s+failed", output)
        if failed_match:
            failed_count = int(failed_match.group(1))
        elif not passed and exit_code != 0:
            failed_count = 1

        error_summary = None
        if not passed:
            error_summary = _extract_error_summary(output)

        return {
            "passed": passed,
            "exit_code": exit_code,
            "output": output,
            "passed_count": passed_count,
            "failed_count": failed_count,
            "error_summary": error_summary,
        }

A few details make this check trustworthy without any LLM opinion:

• tempfile.TemporaryDirectory() gives every run a fresh workspace. The generated code and tests are written there, and the directory is cleaned up when execution ends.
• timeout=15 caps execution time. Code stuck in an infinite loop or running too long gets its subprocess killed and can't block the whole agent workflow.
• exit_code == 0 is the deterministic pass or fail signal from pytest. Zero means the suite succeeded, and any other code means a problem, so the result and stack trace go to the Debugger.
• The results are structured. Rather than returning only raw terminal text, the function pulls out passed and failed counts and builds a short error_summary. That structure lives in the shared state, and the graph uses it to decide between moving ahead and entering the self-healing loop.

Step 4: system prompts for each specialist in src/prompts.py

In a multi-agent system the prompts should be short, authoritative and tightly scoped. Each agent needs a clear picture of its job, its duties and its limits, with no needless instructions and no overlap with its neighbors. The prompts.py file in src defines four distinct system prompts for the specialized roles:

ARCHITECT_SYSTEM_PROMPT = """You are the Lead Software Architect and Quality Assurance Principal.
Your role is to analyze the user's software engineering requirements and design:
1. Architectural Specification: Detailed function/class signatures, data structures, and edge-case requirements.
2. Comprehensive Pytest Suite: A rigorous set of unit tests in `pytest` format.

IMPORTANT TEST RULES:
- The tests MUST import from `solution` (e.g. `from solution import ClassName, function_name`).
- Include test cases for standard happy paths AND tricky edge cases (e.g., empty inputs, boundary limits, invalid types, concurrency/ordering if applicable).
- Write self-contained tests without external network dependencies.

Provide your response in two distinct markdown sections:
### SPECIFICATIONS
<detailed specifications and API contracts>

### TESTS
```python
<complete pytest code>
```
"""

CODER_SYSTEM_PROMPT = """You are a Senior Python Developer.
Your role is to write clean, idiomatic, and robust Python code that implements the provided Architecture Specifications and passes 100% of the Pytest Suite.

RULES:
- Implement ONLY the implementation code (do NOT include tests in your output).
- Use proper typing, error handling, and docstrings.
- Output ONLY the executable Python code inside a ```python ``` code block. Do not include extraneous conversational text.
"""

DEBUGGER_SYSTEM_PROMPT = """You are an Expert Debugger and Causal Root-Cause Analyst.
A Python solution was executed against its test suite in an isolated sandbox and FAILED.

Your job:
1. Carefully inspect the failing Pytest Traceback and Error Summary.
2. Cross-reference the traceback with the provided Implementation Code and Test Suite.
3. Identify the EXACT root cause (e.g., off-by-one error, incorrect data structure, missing edge-case branch).
4. Provide a 2-3 sentence diagnosis explaining the bug.
5. Provide the COMPLETE, corrected Python implementation code.

Output format:
### DIAGNOSIS
<2-3 sentence root-cause diagnosis>

### PATCHED_CODE
```python
<complete corrected python code>
```
"""

REVIEWER_SYSTEM_PROMPT = """You are a Principal Code Reviewer.
The multi-agent engineering team has completed the code and passed all unit tests.
Your job is to provide a concise, high-level code quality summary:
- Verify correctness and adherence to original user requirements.
- Note any key algorithmic strengths or future optimization ideas.
- Keep the summary under 150 words.
"""

The prompts are structured with two goals:

• Explicit output delimiters. ARCHITECT_SYSTEM_PROMPT demands ### SPECIFICATIONS and ### TESTS sections, which lets the parser split the architecture description from the executable test code easily.
• Strict separation of concerns. The Coder is told to write only code and no tests, while the Debugger must give a concise two-sentence diagnosis next to the complete patched code block. Each sub-agent has a clear scope, and that cuts down hallucination.

Step 5: building the sub-agent nodes in src/nodes.py

This is the core execution engine of the workflow, where the node functions get defined. A LangGraph node is a plain Python function. It takes the shared AgentTeamState, performs one specific job such as calling an LLM or running tests, and returns the pieces that need updating. LangGraph merges them into the shared state, so the following node always has the latest context.

Create nodes.py in the src folder and add the nodes one at a time.

Helper parsers for code extraction and spec splitting

Two small utilities keep the pipeline tidy. LLMs often wrap code in Markdown fences, or they open with chatty introductions even when told not to. The helpers handle that formatting so the rest of the pipeline always receives clean, executable Python. The imports and helpers go at the top of src/nodes.py:

import re
from typing import Dict, Any, List
from langchain_core.messages import SystemMessage, HumanMessage

from src.state import AgentTeamState
from src.config import get_llm
from src.sandbox import run_tests_in_sandbox
from src.prompts import (
    ARCHITECT_SYSTEM_PROMPT,
    CODER_SYSTEM_PROMPT,
    DEBUGGER_SYSTEM_PROMPT,
    REVIEWER_SYSTEM_PROMPT,
)


def _get_text(content: Any) -> str:
    """
    Extracts a clean string from either a standard string or list-of-dict content blocks
    returned by modern LangChain message objects.
    """
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for item in content:
            if isinstance(item, str):
                parts.append(item)
            elif isinstance(item, dict) and "text" in item:
                parts.append(str(item["text"]))
        return "\n".join(parts).strip()
    return str(content).strip()


def _extract_code(text: Any) -> str:
    """
    Extracts executable Python code from markdown backticks using regex.
    If no backticks are found, falls back to returning the cleaned raw text.
    """
    clean_text = _get_text(text)
    match = re.search(r"```(?:python)?\s*\n(.*?)\n```", clean_text, re.DOTALL)
    if match:
        return match.group(1).strip()
    return clean_text.strip()


def _parse_architect_output(raw_content: Any) -> tuple[str, str]:
    """
    Splits the Architect's raw output into its two distinct deliverables:
    the architectural specification text and the pytest test suite.
    """
    text = _get_text(raw_content)
    specs = text
    tests = ""

    if "### TESTS" in text:
        parts = text.split("### TESTS")
        specs = parts[0].replace("### SPECIFICATIONS", "").strip()
        tests = _extract_code(parts[1])
    elif "```python" in text:
        tests = _extract_code(text)
        specs = text.split("```python")[0].strip()

    return specs, tests

Each helper covers a common quirk in LLM output:

• _get_text() normalizes the response format. Depending on the provider and LangChain wrapper, a reply can arrive as a plain string or as structured content such as a list with a type and text entry. The helper turns both into an ordinary Python string.
• _extract_code() pulls code out of Markdown fences. It uses a regex with re.DOTALL so multi-line Python is captured whether or not the model wrote python after the opening backticks. When no fenced block exists, it falls back to the cleaned response text.
• _parse_architect_output() splits the Architect's answer at the ### TESTS delimiter. Everything before the marker becomes specs, and everything after it becomes tests. That keeps the natural-language specification apart from the runnable test code handed to the Test Runner later.

The Lead Architect node

The Lead Architect is the team's entry point, and it works test-first. Before any implementation exists, it studies the user's request, defines the data structures and class interfaces, and designs a thorough pytest suite that covers both the expected happy paths and important edge cases. That gives the Developer a clear contract to implement.

The function is added to the same file:

def architect_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 1: Lead Architect (High Reasoning Tier).
    Analyzes user requirements, designs architectural specifications,
    and authors an exhaustive Pytest suite covering happy paths and edge cases.
    """
    llm = get_llm("architect")
    messages = [
        SystemMessage(content=ARCHITECT_SYSTEM_PROMPT),
        HumanMessage(content=f"Requirements:\n{state['user_request']}"),
    ]
    response = llm.invoke(messages)
    specs, tests = _parse_architect_output(str(response.content))

    return {
        "specs": specs,
        "tests": tests,
        "status": "specs_ready",
    }

Three things stand out in the code:

• Model selection. get_llm("architect") sets up gemini-pro-latest with high thinking. A strong model with deep reasoning pays off here, because the work is planning from requirements and foreseeing edge cases like thread safety, empty collections or boundary conditions.
• Message construction. The system prompt ARCHITECT_SYSTEM_PROMPT is wrapped in a SystemMessage, while the raw user prompt from state['user_request'] becomes a HumanMessage.
• State updates. The result is a dictionary holding specs, tests and status = "specs_ready", and LangGraph folds all three keys into AgentTeamState.

The Developer node

Once the specification and tests exist, control passes to the Developer. Its role is to write clean, idiomatic Python that implements the specification and satisfies every test the Architect wrote. The function goes in src/nodes.py as developer_node():

def developer_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 2: Developer / Coder (Fast Workhorse Tier).
    Synthesizes the Python implementation code grounded in the specifications
    and the explicit assertions of the Pytest suite.
    """
    llm = get_llm("coder")
    prompt = f"""Requirements:
{state['user_request']}

Architect Specifications:
{state['specs']}

Pytest Suite:
```python
{state['tests']}
```

Write the complete Python implementation code to satisfy all tests.
"""
    messages = [
        SystemMessage(content=CODER_SYSTEM_PROMPT),
        HumanMessage(content=prompt),
    ]
    response = llm.invoke(messages)
    code = _extract_code(response.content)

    return {
        "code": code,
        "status": "code_generated",
    }

How it works:

• Model selection. get_llm("coder") resolves to gemini-3.7-flash with medium thinking. The Architect already fixed the API blueprint and the test assertions, so the Developer has no need to redesign anything and can concentrate on quick, accurate code. Medium effort is enough for that, and it also keeps token use and development cost down.
• Grounding in the test suite. state['tests'] is injected directly into the Developer's prompt. Seeing the exact assertions it has to pass anchors the model to the expected method names, argument types and return values.
• State updates. After pulling the plain Python out of the reply, the node returns {"code": code, "status": "code_generated"}.

The deterministic test sandbox node

This is one of the system's most important nodes. Rather than asking an LLM whether the code looks right, it runs the pytest suite against the code in an isolated subprocess. The tests run locally, so the step makes zero LLM API calls and usually finishes in about 15 milliseconds.

The test_runner_node() function is added to src/nodes.py:

def test_runner_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 3: Deterministic Test Sandbox ($0.00 / 0 Tokens).
    Executes tests against the code in an isolated temporary subprocess sandbox.
    Returns structured pass/fail metrics and concise error tracebacks upon failure.
    """
    test_results = run_tests_in_sandbox(
        code=state["code"],
        tests=state["tests"],
        timeout_seconds=15,
    )
    new_status = "tests_passed" if test_results["passed"] else "tests_failed"

    return {
        "test_results": test_results,
        "status": new_status,
    }

• Zero-token determinism. run_tests_in_sandbox() receives state["code"] and state["tests"]. No model is called, so this step costs nothing in tokens.
• Status determination. A test_results["passed"] == True outcome (meaning exit_code == 0) sets status to "tests_passed", and anything else sets it to "tests_failed".
• Structured state output. The returned test_results bundles the run output, the counts of passing and failing tests, and a distilled stack-trace summary. This is what the conditional edge inspects during routing.

The self-healing Debugger node

Whenever the sandbox reports failures, control routes to the Debugger, the root-cause specialist. It studies three inputs, namely the broken code, the test assertions and the precise pytest stack trace, to figure out why the run failed, then applies a surgical patch. The debugger_node() function goes into src/nodes.py:

def debugger_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 4: Debugger & Fixer (High Reasoning Tier).
    Performs causal root-cause analysis on execution tracebacks and stack traces.
    Applies surgical code patches and logs the diagnosis to the debug history.
    """
    llm = get_llm("debugger")
    test_results = state.get("test_results", {})
    error_summary = test_results.get("error_summary", test_results.get("output", "Unknown error"))
    current_iteration = state.get("iteration", 0) + 1

    prompt = f"""The implementation failed its unit tests in the execution sandbox.

Current Implementation Code:
```python
{state['code']}
```

Pytest Suite:
```python
{state['tests']}
```

Execution Error Traceback:
{error_summary}

Please analyze the root cause of the failure and output the corrected Python code.
"""
    messages = [
        SystemMessage(content=DEBUGGER_SYSTEM_PROMPT),
        HumanMessage(content=prompt),
    ]
    response = llm.invoke(messages)
    response_text = str(response.content)

    # Extract diagnosis summary and patched code block
    diagnosis = "Debugger applied code patch based on traceback."
    if "### DIAGNOSIS" in response_text and "### PATCHED_CODE" in response_text:
        diag_part = response_text.split("### PATCHED_CODE")[0]
        diagnosis = diag_part.replace("### DIAGNOSIS", "").strip()

    patched_code = _extract_code(response_text)

    # Append the diagnosis to our audit trail
    history = list(state.get("debug_history", []))
    history.append(f"Iteration {current_iteration}: {diagnosis}")

    return {
        "code": patched_code,
        "iteration": current_iteration,
        "debug_history": history,
        "status": "code_patched",
    }

How it behaves:

• High reasoning allocation. get_llm("debugger") gives back gemini-3.7-flash set to high thinking. Reading a stack trace means following the execution path, finding broken edge cases and reasoning about how variables changed, which takes real effort. The model was picked as a compromise between depth and speed. A pro model works as well for this layer, and the flow could even switch models mid-run if the debugger is still stuck after one or two loops.
• Dense error context. Only the trimmed error_summary reaches the model, for instance AssertionError: assert False == True, where False = allow_request('user_1').
• Iteration counting. With current_iteration = state.get("iteration", 0) + 1, each trip through the Debugger moves the retry counter up by one.
• Audit trail. The two-sentence diagnosis the model produced is added to debug_history.
• State updates. The old state['code'] gets swapped for patched_code, and the node reports status = "code_patched".

The Reviewer node

After every test passes, or when the retry limit is hit, control goes to the Reviewer, the final quality gate. It checks overall code quality, confirms the docstrings are clear and useful, and writes a short summary of the finished solution for the user. The reviewer_node() function is added to src/nodes.py:

def reviewer_node(state: AgentTeamState) -> Dict[str, Any]:
    """
    Sub-Agent 5: Code Reviewer (Fast Mode Tier).
    Validates code quality, verifies requirement adherence,
    and produces the final executive deliverable summary.
    """
    llm = get_llm("reviewer")
    test_results = state.get("test_results", {})

    prompt = f"""Requirements:
{state['user_request']}

Final Implementation Code:
```python
{state['code']}
```

Test Results:
Passed: {test_results.get('passed', False)} ({test_results.get('passed_count', 0)} passed, {test_results.get('failed_count', 0)} failed)
Iterations Taken: {state.get('iteration', 0)}
Debug History: {state.get('debug_history', [])}

Provide a concise, professional code quality review and verification summary.
"""
    messages = [
        SystemMessage(content=REVIEWER_SYSTEM_PROMPT),
        HumanMessage(content=prompt),
    ]
    response = llm.invoke(messages)

    return {
        "review_summary": _get_text(response.content),
        "status": "completed",
    }

What it does:

• Model selection. get_llm("reviewer") yields gemini-flash-lite-latest at a low thinking level, suited to quick summary writing. Little reasoning is needed at this stage, so a lighter model with lower thinking capacity gives better speed and cost.
• Contextual review. The Reviewer receives the final implementation, the pass and fail figures from the sandbox, the number of debug iterations used and the complete debug history.
• State updates. The node hands back {"review_summary": _get_text(response.content), "status": "completed"}.

Step 6: routing logic in src/edges.py

The core logic and agent nodes are done, so routing is next. Nodes perform the computation, and edges decide where control goes. A conditional edge acts as a dynamic router: it examines the state dictionary a node returned and answers with the name of the next node to execute.

The routing function lives in a new edges.py file inside src:

from typing import Literal
from src.state import AgentTeamState

def route_after_test_runner(state: AgentTeamState) -> Literal["reviewer", "debugger"]:
    """
    Conditional edge router evaluated immediately after `test_runner` completes.
    Decides whether to trigger the self-healing Debugger loop or proceed to Reviewer.
    """
    test_results = state.get("test_results", {})
    passed = test_results.get("passed", False)
    iteration = state.get("iteration", 0)
    max_iterations = state.get("max_iterations", 3)

    # 1. Happy Path: 100% of unit tests passed -> Proceed directly to Review
    if passed:
        return "reviewer"

    # 2. Self-Healing Loop: Tests failed and within retry budget -> Trigger Debugger
    if iteration < max_iterations:
        return "debugger"

    # 3. Safety Fallback: Exceeded maximum retries -> Exit loop to Reviewer with warning
    return "reviewer"

The router looks at the test results and the number of debugging attempts so far, and it handles three cases:

• Happy path (passed == True). When every test passes on the first attempt, the workflow goes straight to the Reviewer, and there's no reason to spend extra time or tokens on the Debugger.
• Self-healing (passed == False while iteration < max_iterations). Failing tests with attempts still available send the code to the Debugger, which analyzes the failure and tries a fix.
• Retry limit reached (iteration >= max_iterations). When the Debugger has burned all three permitted attempts and tests still fail, the workflow goes to the Reviewer rather than looping without end. This hard ceiling on self-healing stops the graph from stalling on contradictory or ambiguous requirements.

Step 7: assembling the graph in src/graph.py

Nodes and routing are done, so now everything gets connected. Direct handoffs and conditional feedback edges link the nodes, and the result compiles into a StateGraph that can run the whole workflow. Put this code in a new graph.py file inside src:

from langgraph.graph import StateGraph, END
from src.state import AgentTeamState
from src.nodes import (
    architect_node,
    developer_node,
    test_runner_node,
    debugger_node,
    reviewer_node,
)
from src.edges import route_after_test_runner

def build_self_healing_graph() -> StateGraph:
    """
    Assembles the Multi-Agent Self-Healing Engineering Team into a LangGraph StateGraph.
    Configures node registrations, linear handoffs, cyclic feedback edges, and conditional routing.
    """
    # 1. Initialize StateGraph with our Shared Whiteboard State Schema
    workflow = StateGraph(AgentTeamState)
    # 2. Register Sub-Agent Nodes (Mapping node names to Python callable functions)
    workflow.add_node("architect", architect_node)
    workflow.add_node("developer", developer_node)
    workflow.add_node("test_runner", test_runner_node)
    workflow.add_node("debugger", debugger_node)
    workflow.add_node("reviewer", reviewer_node)
    # 3. Define Graph Entry Point
    workflow.set_entry_point("architect")
    # 4. Add Deterministic Direct Handoff Edges
    workflow.add_edge("architect", "developer")
    workflow.add_edge("developer", "test_runner")

    # 5. Add the Cyclic Self-Healing Feedback Edge
    workflow.add_edge("debugger", "test_runner")
    # 6. Add Conditional Routing Edge for Sandbox Test Evaluation
    workflow.add_conditional_edges(
        "test_runner",
        route_after_test_runner,
        {
            "debugger": "debugger",
            "reviewer": "reviewer",
        },
    )
    # 7. Add Termination Edge
    workflow.add_edge("reviewer", END)
    # 8. Compile the Graph into an Executable State Machine
    return workflow.compile()

The construction goes step by step:

1. StateGraph(AgentTeamState) creates a stateful graph using the AgentTeamState schema, so every node can reach the same shared state as the workflow advances.
2. workflow.add_node(name, func) registers each Python function as a node. Each has a single responsibility: designing tests, generating code, running tests or debugging failures.
3. workflow.set_entry_point("architect") fixes the starting point. Invoking or streaming the graph with an initial state begins execution at the architect node.
4. workflow.add_edge("architect", "developer") and workflow.add_edge("developer", "test_runner") build the initial straight line. When the Architect has produced the specification and tests, the Developer takes over, and when the Developer has produced code, the Test Runner follows.
5. workflow.add_edge("debugger", "test_runner") closes the self-healing loop, which turns a plain pipeline into a self-healing system. A fixed bug doesn't send the workflow on to the Reviewer. The patched code returns to the Test Runner so the fix gets validated again.
6. workflow.add_conditional_edges("test_runner", route_after_test_runner, ...) makes LangGraph call route_after_test_runner(state) once tests finish. Its return value sends the workflow to either the debugger or the reviewer.
7. workflow.add_edge("reviewer", END) ends the run. After the Reviewer completes its quality check and summary, the graph execution is over.
8. workflow.compile() turns the graph definition into a runnable graph, validating the structure and preparing it for execution.

Step 8: the CLI entry point in src/main.py

The last piece is a command-line entry point built with rich, which streams each sub-agent's progress in real time using styled terminal panels. The main.py file in src begins with the banner and configuration-table display functions:

import os
import sys
import argparse
import time
import warnings

# Suppress internal SDK library warnings
warnings.filterwarnings("ignore")
# Ensure project root is in sys.path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.syntax import Syntax
from src.config import DEFAULT_CONFIG, MAX_DEBUG_ITERATIONS
from src.graph import build_self_healing_graph
from src.state import AgentTeamState
console = Console()

def print_banner():
    console.print(
        Panel.fit(
            "[bold cyan]LangGraph Self-Healing Multi-Agent Engineering Team[/bold cyan]\n"
            "[dim]Heterogeneous Model Tiering | Sub-Agent Tool Scoping | Autonomous Pytest Sandbox[/dim]",
            border_style="cyan",
        )
    )

def print_config_table():
    table = Table(title="Multi-Agent Team Configuration and Model Tiers", border_style="dim")
    table.add_column("Sub-Agent Role", style="bold yellow")
    table.add_column("Assigned Model", style="cyan")
    table.add_column("Thinking Level", style="green")
    table.add_column("Primary Responsibility", style="white")
    table.add_row(
        "Architect",
        DEFAULT_CONFIG["architect"]["model"],
        str(DEFAULT_CONFIG["architect"]["thinking_level"]).upper(),
        "Specifications and Pytest Edge-Case Suite",
    )
    table.add_row(
        "Developer",
        DEFAULT_CONFIG["coder"]["model"],
        str(DEFAULT_CONFIG["coder"]["thinking_level"]).upper(),
        "Fast Code Synthesis from Specification",
    )
    table.add_row(
        "Test Runner",
        "Deterministic Python ($0.00 / 0 Tokens)",
        "N/A",
        "Subprocess Pytest Sandbox Execution",
    )
    table.add_row(
        "Debugger",
        DEFAULT_CONFIG["debugger"]["model"],
        str(DEFAULT_CONFIG["debugger"]["thinking_level"]).upper(),
        "Causal Traceback Analysis and Patching",
    )
    table.add_row(
        "Reviewer",
        DEFAULT_CONFIG["reviewer"]["model"],
        str(DEFAULT_CONFIG["reviewer"]["thinking_level"]).upper(),
        "Code Verification and Executive Summary",
    )
    console.print(table)
    console.print()

These are purely about terminal styling and are self-explanatory, so they don't need a deeper walkthrough. The same file continues with run_agent_pipeline():

def run_agent_pipeline(user_request: str):
    print_banner()
    print_config_table()

console.print(Panel(f"[bold white]User Request:[/bold white]\n{user_request}", border_style="blue"))
    initial_state: AgentTeamState = {
        "user_request": user_request,
        "specs": "",
        "tests": "",
        "code": "",
        "test_results": {},
        "iteration": 0,
        "max_iterations": MAX_DEBUG_ITERATIONS,
        "debug_history": [],
        "review_summary": "",
        "status": "starting",
    }
    graph = build_self_healing_graph()
    start_time = time.time()
    current_state = initial_state
    console.print("[bold green]Starting LangGraph Multi-Agent Workflow...[/bold green]\n")
    # Stream graph execution node by node
    for event in graph.stream(initial_state):
        for node_name, node_output in event.items():
            current_state.update(node_output)
            if node_name == "architect":
                console.print(
                    Panel(
                        f"[bold yellow][1/5] Architect Completed Specifications & Tests[/bold yellow]\n\n"
                        f"[bold]Specifications:[/bold]\n{current_state['specs'][:300]}...\n\n"
                        f"[bold]Generated Pytest Suite:[/bold]\n[dim]({len(current_state['tests'].splitlines())} lines of test code designed)[/dim]",
                        border_style="yellow",
                    )
                )
            elif node_name == "developer":
                console.print(
                    Panel(
                        f"[bold cyan][2/5] Developer Completed Initial Implementation[/bold cyan]\n"
                        f"[dim]Synthesized {len(current_state['code'].splitlines())} lines of Python code.[/dim]",
                        border_style="cyan",
                    )
                )
            elif node_name == "test_runner":
                results = current_state["test_results"]
                if results.get("passed", False):
                    console.print(
                        Panel(
                            f"[bold green][3/5] Test Runner Sandbox: PASSED[/bold green]\n"
                            f"[green]All {results.get('passed_count', 0)} unit tests passed cleanly in subprocess sandbox.[/green]",
                            border_style="green",
                        )
                    )
                else:
                    console.print(
                        Panel(
                            f"[bold red][3/5] Test Runner Sandbox: TESTS FAILED[/bold red]\n"
                            f"[red]{results.get('failed_count', 0)} failed, {results.get('passed_count', 0)} passed.[/red]\n\n"
                            f"[bold white]Traceback Snippet:[/bold white]\n[dim red]{results.get('error_summary', '')[:400]}[/dim red]",
                            border_style="red",
                        )
                    )
            elif node_name == "debugger":
                history = current_state.get("debug_history", [])
                latest_diag = history[-1] if history else "Patch applied."
                console.print(
                    Panel(
                        f"[bold magenta][4/5] Self-Healing Debugger (Iteration {current_state.get('iteration', 1)})[/bold magenta]\n"
                        f"[magenta]{latest_diag}[/magenta]\n\n"
                        f"[dim]Re-routing patched code to Test Runner Sandbox...[/dim]",
                        border_style="magenta",
                    )
                )
            elif node_name == "reviewer":
                console.print(
                    Panel(
                        f"[bold white][5/5] Reviewer Final Report[/bold white]\n"
                        f"{current_state.get('review_summary', '')}",
                        border_style="white",
                    )
                )
    elapsed = round(time.time() - start_time, 2)
    # Display Final Deliverables
    console.print("\n" + "=" * 80)
    console.print(f"[bold green]Pipeline Completed in {elapsed}s[/bold green]\n")
    console.print("[bold cyan]Final Verified Implementation (solution.py):[/bold cyan]")
    syntax_code = Syntax(current_state["code"], "python", theme="monokai", line_numbers=True)
    console.print(syntax_code)
    console.print("\n[bold yellow]Comprehensive Pytest Suite (test_solution.py):[/bold yellow]")
    syntax_tests = Syntax(current_state["tests"], "python", theme="monokai", line_numbers=True)
    console.print(syntax_tests)
    # Summary Metrics Table
    summary_table = Table(title="Execution Summary Metrics", border_style="cyan")
    summary_table.add_column("Metric", style="bold")
    summary_table.add_column("Value", style="green")
    summary_table.add_row("Total Elapsed Time", f"{elapsed} seconds")
    summary_table.add_row("Self-Healing Iterations", str(current_state.get("iteration", 0)))
    summary_table.add_row("Pytest Pass Status", "Passed (100%)" if current_state["test_results"].get("passed") else "Failed")
    summary_table.add_row("Unit Tests Passed", str(current_state["test_results"].get("passed_count", 0)))
    console.print(summary_table)

run_agent_pipeline() is where the pieces meet. It sets up the shared AgentTeamState, builds the self-healing graph and streams execution node by node. After each node completes, it refreshes the current state and prints progress in Rich panels: the Architect's specification and tests, the Developer's code, the Test Runner's outcome, each debugging round and the Reviewer's closing report. At the end it prints the verified implementation, the generated pytest suite and run metrics such as total time, debugging iterations and test results.

The CLI argument parser finishes the file:

def main():
    parser = argparse.ArgumentParser(description="LangGraph Self-Healing Multi-Agent Engineering Team")
    parser.add_argument(
        "--prompt",
        type=str,
        default="Build an in-memory Sliding Window Rate Limiter class in Python with thread safety, per-user limits, and a clean method `allow_request(user_id: str) -> bool`.",
        help="Software engineering prompt for the multi-agent team",
    )
    args = parser.parse_args()
    run_agent_pipeline(args.prompt)

if __name__ == "__main__":
    main()

Running it on a rate limiter

For a real test, the team gets a non-trivial algorithmic task. This command asks for a thread-safe sliding window rate limiter:

python src/main.py --prompt "Build an in-memory Sliding Window Rate Limiter class in Python with thread safety, per-user limits, and a method allow_request(user_id: str) -> bool."

The terminal output shows the following sequence:

1. Stage 1 of 5, the Lead Architect, analyzes the rate limiter requirements and produces a multi-threaded pytest suite covering timestamp pruning, sliding window threshold enforcement and concurrent requests from many threads.
2. Stage 2 of 5, the Developer, writes a first SlidingWindowRateLimiter class built on collections.deque and a threading.Lock.
3. Stage 3 of 5, the Test Runner on iteration 0, runs pytest. The concurrency stress test fails because timestamp pruning happened outside the locked critical section, so a race condition let several threads exceed the request allowance.
4. Stage 4 of 5, the Debugger, reads the traceback and reports that the rate limiter failed the concurrent thread test because _prune_expired_timestamps() was called before self._lock was acquired, and that the fix wraps the timestamp eviction in the synchronized block.
5. Stage 3 once more, the Test Runner on iteration 1, runs pytest on the patched code, and the suite goes green with 6 passed in 0.04s.
6. Stage 5 of 5, the Reviewer, summarizes the thread-safe implementation and the verified test metrics, and then delivers the final result.

A screenshot of the real terminal output from this run:

The full output also includes the Python code that was written and the test cases used to check it. Reproducing all of it here would add length without much value.

Where the pattern applies

Combining specialized sub-agents, a different model and thinking level for each, and a deterministic execution sandbox turns a fragile single-prompt workflow into a more dependable, self-healing system.

The demonstration was a code generation team, but the same state-machine pattern fits many other workflows, such as data pipelines, dynamic API orchestration and structured document validation. The core idea is plain: don't depend on one model to do everything. Give each agent a focused responsibility and let the system check and correct its own work as it goes.

Three lessons stand out:

1. Never let an LLM grade its own code. Hand execution to a local deterministic subprocess sandbox running pytest, which is fast, free and gives an exact pass or fail answer.
2. Match the thinking level to the role. Architecture and debugging deserve high reasoning, balanced code synthesis fits medium, and quick summary writing can run on low.
3. Cyclic feedback loops are where AI engineering is heading. State machines that correct themselves from real runtime feedback should beat static prompt chains by a wide margin.