告别单体AI缺陷!LangGraph分层自愈多智能体实战搭建

来源：DeepHub IMBA
本文约13000字，建议阅读15+分钟
本文介绍了基于LangGraph搭建分层自愈多智能体系统的完整实战方案。

编排专业化 AI 子智能体：按角色分配思考级别、隔离工具沙箱，并用循环反馈构建更可靠的 Python 智能体系统

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

 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

 pip install -r requirements.txt

 # API Key GEMINI_API_KEY="your_api_key_here" # OPENROUTER_API_KEY="" # OPENAI_API_KEY="" # 模型分层选择与自适应思考级别（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" # 自愈式重试循环的最大次数 MAX_DEBUG_ITERATIONS=3

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

 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

 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() # 角色配置与自适应思考级别 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"))

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（处理 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. 直接使用 OpenAI Provider 作为回退方案     if openai_key:         return ChatOpenAI(             model=model_name,             api_key=openai_key,             temperature=temp,         )     return ChatGoogleGenerativeAI(model=model_name, temperature=temp)

import osimport reimport sysimport tempfileimport subprocessfrom typing import Dict, Anydef _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()    # 回退到最后 25 行    lines = pytest_output.splitlines()    return "\n".join(lines[-25:])

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")        # 把解决方案代码和测试文件写入临时目录        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)        # 使用当前 Python 解释器构建执行命令        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,        }

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."""

import refrom typing import Dict, Any, Listfrom langchain_core.messages import SystemMessage, HumanMessagefrom src.state import AgentTeamStatefrom src.config import get_llmfrom src.sandbox import run_tests_in_sandboxfrom 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

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",    }

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",    }

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,    }

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)    # 提取诊断摘要和修复后的代码块    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)    # 把诊断结果追加到审计记录    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",    }

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",    }

from typing import Literalfrom src.state import AgentTeamStatedef 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. 正常路径：100% 的单元测试通过 -> 直接进入审查    if passed:        return "reviewer"    # 2. 自愈循环：测试失败且仍在重试预算内 -> 触发调试器    if iteration < max_iterations:        return "debugger"    # 3. 安全回退：超过最大重试次数 -> 带警告退出循环并进入审查器    return "reviewer"

from langgraph.graph import StateGraph, ENDfrom src.state import AgentTeamStatefrom src.nodes import (    architect_node,    developer_node,    test_runner_node,    debugger_node,    reviewer_node,)from src.edges import route_after_test_runnerdef 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. 使用共享白板状态 Schema 初始化 StateGraph    workflow = StateGraph(AgentTeamState)    # 2. 注册子智能体节点（把节点名称映射到 Python 可调用函数）    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. 定义图的入口点    workflow.set_entry_point("architect")    # 4. 添加确定性的直接交接边    workflow.add_edge("architect", "developer")    workflow.add_edge("developer", "test_runner")    # 5. 添加循环式自愈反馈边    workflow.add_edge("debugger", "test_runner")    # 6. 为沙箱测试评估添加条件路由边    workflow.add_conditional_edges(        "test_runner",        route_after_test_runner,        {            "debugger": "debugger",            "reviewer": "reviewer",        },    )    # 7. 添加终止边    workflow.add_edge("reviewer", END)    # 8. 把图编译成可执行状态机    return workflow.compile()

import osimport sysimport argparseimport timeimport warnings# 抑制 SDK 库内部警告warnings.filterwarnings("ignore")# 确保项目根目录位于 sys.path 中sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))from rich.console import Consolefrom rich.panel import Panelfrom rich.table import Tablefrom rich.syntax import Syntaxfrom src.config import DEFAULT_CONFIG, MAX_DEBUG_ITERATIONSfrom src.graph import build_self_healing_graphfrom src.state import AgentTeamStateconsole = 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()

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")    # 逐节点流式执行图    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)    # 显示最终交付物    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_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)

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

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."