LangGraph中的自修复多智能体团队：架构师、测试员、调试员
用 Python 构建一个 LangGraph 状态机，让架构师、开发人员、pytest 沙箱环境、调试器以及审查员协同工作并循环迭代，直至测试通过。
如今，人工智能几乎渗透到了软件开发的各个层面，从代码生成到数据分析，再到多步骤推理。然而，任何向大型语言模型请求复杂任务的人都会遇到同样的问题：表面上看似乎正确的答案，在面对实际需求及一些边缘情况时就会失效。通常的解决办法是手动循环操作——复制错误日志，粘贴到聊天界面中，请求修复，如此反复直到问题最终解决。
另一种方法也有其弊端。将数十种工具和一长串指令交给同一个整体式智能体，往往会提高错误率，因为其上下文会被污染，各工具的功能特征也会变得模糊不清。由于智能体无法同时执行所有指令，输出质量也会下降。
下面的设计通过使用 LangGraph 用 Python 编写的模块化、分层的多智能体状态机解决了这两个问题。该示例展示了一个具备自我修复能力的软件工程团队，由五个角色组成：架构师、开发人员、基于子进程的 Pytest 沙箱、调试器以及审查员。这种模式也可应用于许多其他人工智能工程工作流中。
将大型任务分解给专门的子智能体，让每个智能体承担适当的推理工作，在受控的沙箱环境中运行结果，并加入持续的反馈循环，从而使团队能够测试自己的工作，在交付经过验证的结果之前发现并修复问题。首先，让我们来看看其架构。
该系统由两个原则构成。
首先是具有自适应思维级别的异构模型分层。不同角色的任务难度各异，因此不应都使用相同的模型或相同的计算资源：
- 架构师使用
gemini-pro-latest模型，并配备高思维级别。它负责处理最需要推理的工作，比如确定问题边界、预判竞态条件以及设计严格的pytest测试套件。 - 开发人员使用
gemini-3.7-flash模型，并配备中思维级别。这种配置能在遵循架构师明确要求的的同时，实现快速的代码生成。 - 测试运行器是一个确定性的Python子进程，可在大约15毫秒内本地运行
pytest，且完全不消耗LLM API令牌。
gemini-3.7-flash 并启用 high 级思考模式，因为要追溯原始堆栈跟踪信息至出错行，必须进行因果根本原因分析。gemini-flash-lite-latest 并启用 low 级思考模式。检查代码风格以及为用户撰写总结仅需简单的推理即可完成。第二个原则是严格的工具范围限制，即对各个智能体实施最小权限原则。没有任何子智能体可以无限制地访问终端和文件系统，每个智能体只能获得其角色所需的输入和工具。开发者无法运行任意的shell命令，而测试运行器仅负责执行测试并报告实际发生的情况，不会要求大型语言模型来解释或编造结果。严格的限制能让每个智能体保持专注，从而减少幻觉现象。
涉及到的LangGraph概念
LangGraph 的运作方式类似于共享白板。每位团队成员，也就是每个子智能体，都会执行分配给自己的任务，并将结果写回同一个共享空间，这样所有人都能看到团队目前构建的上下文。LangGraph 则充当协调者：它决定下一个由哪个智能体执行任务，并传递相关的状态信息。
这里有四个关键概念：
- 状态，即
TypedDict，就是那个白板。所有智能体都读取并写入一个名为AgentTeamState的 Python 字典，该字典中存储着用户提示、规范、单元测试、当前的代码草稿、测试结果以及调试轮次数。 - 节点对应团队成员。节点其实只是一个 Python 函数，它接收当前状态，执行相应任务（如调用大型语言模型、运行测试），然后返回更新内容，这些更新会被合并回共享状态中。
StateGraph注册节点和边并完成编译后得到的结果。该图结构可以被运行，且在工作流推进过程中其事件可以被实时传输。项目设置
首先创建一个全新的目录和一个虚拟环境。以下终端命令可用于创建这两个元素并激活虚拟环境：
mkdir langgraph-self-healing-team
cd langgraph-self-healing-team
python3 -m venv .venv
source .venv/bin/activate
requirements.txt 文件会列出所有依赖项：
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
每个组件都有其职责：
langgraph负责协调多智能体工作流程，处理状态管理、节点间的数据传递以及循环反馈机制。langchain-google-genai用于连接 Google Gemini 模型，并展示它们的思考过程。langchain-openai是可选组件。它提供了对 OpenRouter 或 OpenAI 的支持，这样在不想仅依赖 Gemini 时，就可以使用 GPT 5.6 或 Deepseek v4 Pro 等其他模型。pytest是本地子进程沙箱中的测试运行工具，用于执行生成的测试套件并返回实际结果。rich能够通过颜色、表格和语法高亮来格式化终端输出，从而让智能体工作流程更易于理解和调试。python-dotenv可以自动从.env文件中加载 API 密钥和模型设置。
项目根目录中的 .env 文件用于存储凭证和模型配置：
# 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
环境准备就绪后，就可以逐步构建应用程序了。
第一步：在 src/state.py 中定义共享状态
首先需要明确各个组件之间传递数据的格式。创建一个 src 目录，并在其中放入一个 state.py 文件。
第一个定义是一个使用 Pydantic 编写的 TestResult 模型，它代表了沙箱环境中 pytest 运行后的结构化结果：
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用于记录测试套件是否通过、进程退出码、通过与失败的测试数量以及原始的测试输出。它还包含一个error_summary字段，其中仅存放堆栈跟踪中的相关部分，调试器可借此定位并修复故障。
接下来是作为TypedDict存在的核心AgentTeamState。它就像在图中移动的白板：每个节点读取所需信息，执行相应操作，并更新后续节点所依赖的字段。
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
每个字段在工作流中都承担着特定角色：
user_request存储用户提出的原始问题，这样在整个流程中所有处理节点都能看到最初的意图。specs和tests由架构师提供。开发人员根据这些规范进行实现，而测试运行器则针对代码执行生成的测试。code是当前的 Python 实现代码。开发人员首先填写该内容，若运行失败，调试器可对其进行替换。test_results保存来自沙箱的结构化测试结果，这些结果决定了由审核员还是调试器来处理问题。iteration和max_iterations用于跟踪自我修复循环。它们可防止调试器无限重试，并设定安全上限，例如最多尝试三次。
debug_history 会记录每次调试的根本原因及解决方案，以及代码的变更历史。review_summary 会在代码通过验证后接收审核人的最终报告。status 显示工作流的当前状态：正在生成代码、运行测试、调试中或已准备好审核。第2步：src/config.py中的模型层级与思维级别
统一的LLM工厂会为每个子代理角色分配合适的模型及推理级别（低、中或高）。文件 src/config.py 以一个默认的配置字典开头：
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"))
每个角色都会获得与其任务难度相匹配的模型和思考层级，从而在推理能力、速度与成本之间取得平衡，同时不影响整体质量：
architect和debugger的层级被设置为high，因为系统设计和堆栈跟踪诊断都需要深入的多步骤推理。coder的层级被设置为medium。它只需按照架构师制定的计划执行，因此具备均衡、快速且扎实的综合能力即可。reviewer的层级被设置为low以节省算力，因为它所做的只是查看结果后撰写总结。
同一个文件中还包含了get_llm()工厂函数：
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)
这个函数之所以有用，原因有二。其一，它会将thinking_level（low、medium、high）直接传递给ChatGoogleGenerativeAI，从而可以根据不同角色调整推理强度。其二，它返回的所有客户端都实现了LangChain的BaseChatModel，因此无论某个角色背后使用的是哪种服务提供商，图节点调用.invoke()的方式都保持一致。
第3步：在src/sandbox.py中创建确定性的pytest沙箱
在智能体设计中常见的错误是让大语言模型自行验证它刚刚编写的代码。即便代码中存在缺陷，模型也可能会自信地预测代码通过测试，因为这类模型常常会误以为自己的输出是正确的。即便模型确实检测到了一些错误，也无法保证它已经找出了所有缺陷。
一种更可靠的方案是让代码自行证明其可靠性。无需依赖模型的判断，本地执行节点会将生成的代码及测试用例写入临时目录，然后在独立的 Python 子进程中运行 pytest。这样就能得到客观的通过或失败结果，可直接反馈到工作流中。
在 src 文件夹中创建 sandbox.py 文件。该文件首先定义了一个辅助函数 _extract_error_summary()，用于从 pytest 的原始输出中提取最相关的错误细节，这样调试器就能看到有用的错误追踪信息，而无需查看全部输出内容：
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:])
该辅助工具会去除终端颜色代码以及样板测试标题，仅保留具体的断言失败信息（assert result == expected）以及堆栈跟踪信息。这样一来，调试器就能快速获得关于问题所在的高效描述，而不会产生不必要的冗余内容。
主运行函数 run_tests_in_sandbox() 也位于同一个文件中：
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,
}
一些细节使得无需依赖任何大语言模型的意见，就能确保此检查的可靠性：
tempfile.TemporaryDirectory()为每次运行提供全新的工作空间。生成的代码和测试文件都保存在此处，执行结束后该目录会被清理。timeout=15可限制执行时间。陷入无限循环或运行过长的代码其子进程会被终止，从而不会阻塞整个代理工作流程。exit_code == 0是pytest返回的确定性的通过或失败信号。值为零表示测试套件成功，其他数值则表示存在问题，因此结果和堆栈跟踪信息会被发送到调试器。
error_summary。这一结构存储在共享状态中，图表会依据它来决定是继续执行还是进入自愈循环。第4步：在src/prompts.py中为每个专家编写系统提示
在多智能体系统中，提示应简短、权威且范围明确。每个智能体都需要清晰了解自己的任务、职责和限制，避免不必要的指令以及与相邻智能体的功能重叠。src目录下的prompts.py文件为不同专业角色定义了四种不同的系统提示：
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.
"""
这些提示语的构建基于两个目标：
- 明确的输出分隔符。
ARCHITECT_SYSTEM_PROMPT要求包含### SPECIFICATIONS和### TESTS两部分，这样解析器就能轻松将架构描述与可执行的测试代码分开。 - 严格的职责划分。编码器只需编写代码，不得包含测试内容；而调试器则必须在完整的修复代码块旁给出简明的两句话诊断。每个子代理都有明确的职责范围，从而减少错误生成。
第5步：在src/nodes.py中构建子代理节点
这是工作流的核心执行引擎，节点功能都在这里定义。LangGraph的节点其实就是普通的Python函数。它接收共享的AgentTeamState，执行特定任务（如调用大语言模型或运行测试），然后返回需要更新的数据。LangGraph会将这些数据合并到共享状态中，这样后续的节点总能获得最新的上下文。
在src文件夹中创建nodes.py文件，然后逐个添加节点。
用于代码提取和规范拆分的辅助解析器
有两个小型工具帮助保持处理流程的整洁。大语言模型常常会将代码包裹在Markdown代码块中，或者即便被要求不要这样做也会加上冗长的介绍文字。这些辅助工具会处理这类格式问题，确保后续流程始终能收到干净、可直接运行的Python代码。导入语句和辅助函数放在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
每个辅助函数针对大语言模型输出中的常见问题：
_get_text()用于标准化响应格式。根据提供方及LangChain封装方式，回复可能以普通字符串形式出现，也可能以包含type和text字段的列表等结构化内容呈现。该辅助函数会将这两种形式统一转换为普通的Python字符串。_extract_code()用于从Markdown代码块中提取代码。它使用带有re.DOTALL标志的正则表达式，因此无论模型是否在开头反引号后写入“python”，都能捕获多行Python代码。若不存在代码块，则会回退到已清理过的响应文本。
_parse_architect_output()会以### TESTS作为分隔符来拆分架构师的回答。分隔符之前的内容被归类为specs，之后的内容则被归类为tests。这样就能将自然语言形式的规范与日后交给测试运行器的可执行测试代码区分开来。首席架构师节点
首席架构师是团队的起点，采用以测试为先的工作方式。在任何实现代码出现之前，它会先研究用户的需求，定义数据结构与类接口，进而设计一套完善的pytest测试套件，涵盖所有预期的正常流程以及重要的边界情况。这样开发者就能有一个明确的实现依据。
该函数被添加到同一个文件中：
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",
}
代码中有三点值得注意：
- 模型选择。
get_llm("architect")会选用具备high思维能力的gemini-pro-latest模型。由于需要从需求出发进行规划，并考虑线程安全、空集合或边界条件等边缘情况，因此强大的模型与深度推理能力在这里显得尤为重要。 - 消息构建。系统提示语
ARCHITECT_SYSTEM_PROMPT会被封装在SystemMessage中，而来自state['user_request']的原始用户提示语则会被转换为HumanMessage。 - 状态更新。最终会生成一个包含
specs、tests以及status = "specs_ready"的字典，LangGraph会将这三个键合并到AgentTeamState中。
开发者节点
一旦有了规范和测试用例，控制权就会交给开发人员。他的任务是编写简洁、符合Python惯用的代码来实现这些规范，并通过架构师设计的所有测试。该函数将被保存在src/nodes.py文件中，命名为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",
}
工作原理：
- 模型选择。
get_llm(\"coder\")会选定gemini-3.7-flash模型，并设置medium级别的思考模式。架构师已预先设计好API框架和测试断言，因此开发者无需重新设计任何内容，只需专注于编写快速且准确的代码即可。中等强度的配置已足够实现这一目标，同时还能降低Token消耗和开发成本。 - 基于测试套件的引导。
state['tests']会直接被注入到开发者的提示语中。开发者能够看到必须通过的精确断言，这有助于模型牢记预期的方法名、参数类型及返回值。 - 状态更新。在从回复中提取出纯Python代码后，该节点会返回
{"code": code, "status": "code_generated"}。
确定性测试沙箱节点
这是系统中最重要的节点之一。它不会询问大型语言模型代码是否正确，而是会在一个独立的子进程中运行pytest测试套件。这些测试在本地执行，因此不会发起任何大型语言模型API调用，通常只需约15毫秒即可完成。
test_runner_node()函数被添加到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,
}
- 零令牌确定性。
run_tests_in_sandbox()接收state[“code”]和state[“tests”]。由于不会调用任何模型，因此此步骤不消耗任何令牌。 - 状态判定。
test_results[“passed”] == True的结果（即exit_code == 0）会将status设置为“tests_passed”，其他情况则将其设置为“tests_failed”。 - 结构化状态输出。返回的
test_results包含了运行结果、通过与失败的测试数量，以及简化的堆栈跟踪摘要。路由时条件分支会检查这些内容。
自修复调试器节点
每当沙箱报告失败时，控制流程会转向调试器——这一专门处理根本原因的工具。它会分析三个输入信息，即出错的代码、测试断言以及精确的 pytest 堆栈跟踪信息，从而找出运行失败的原因，随后进行精准修复。debugger_node() 函数位于 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",
}
其运行方式：
- 较高的推理资源分配。
get_llm("debugger")会将gemini-3.7-flash的思考模式设置为high。读取堆栈跟踪需要追踪执行路径，找出异常的边界情况，并分析变量变化的原因，这需要耗费大量精力。该模型是在深度与速度之间折中选择的。对于这一层任务，专业级模型同样适用，而且如果在一到两次循环后调试器仍无法继续工作，系统甚至可以在运行过程中切换模型。 - 丰富的错误上下文。只有经过简化的
error_summary会被传递给模型，例如AssertionError: assert False == True, where False = allow_request('user_1')。
current_iteration = state.get("iteration", 0) + 1，每次遍历调试器时重试计数器都会加一。debug_history 中。state['code'] 会被替换为 patched_code，同时节点会报告 status = "code_patched"。审查节点
每次测试通过后，或达到重试次数限制时，控制权会转移到作为最终质量关卡的审查节点。它会检查代码的整体质量，确认文档字符串清晰且有用，并为用户撰写已完成解决方案的简短总结。reviewer_node() 函数被添加到 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",
}
其功能如下：
- 模型选择。
get_llm("reviewer")会返回思维层级为low的gemini-flash-lite-latest模型，适合快速撰写摘要。此阶段几乎不需要推理能力，因此使用思维能力较弱的轻量级模型能提升速度并降低成本。 - 上下文审查。审查者会收到最终实现代码、沙箱测试的通过与失败次数、调试迭代次数以及完整的调试历史记录。
- 状态更新。该节点会返回
{"review_summary": _get_text(response.content), "status": "completed"}。
第6步：src/edges.py中的路由逻辑
核心逻辑和智能体节点已经完成，接下来是路由功能。节点负责执行计算，而边则决定控制流的方向。条件边充当动态路由器：它会查看节点返回的状态字典，并给出下一个要执行的节点名称作为响应。
路由功能位于src目录下的新文件edges.py中：
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"
路由器会查看测试结果以及迄今为止的调试尝试次数，进而处理三种情况：
- 正常流程（
passed == True）。当所有测试在首次尝试时就通过时，工作流会直接传递给审核者，无需再为调试器投入额外时间或资源。 - 自动修复（
passed == False且iteration < max_iterations）。那些在还有尝试次数时失败的测试会被发送给调试器，由其分析故障并尝试修复。 - 达到重试上限（
iteration >= max_iterations）。当调试器用尽了所有允许的三次尝试后测试仍失败时，工作流会转向审核者，而不会无限循环。这种对自动修复设定的硬性上限可防止流程因相互矛盾或模糊的需求而停滞。
第7步：在src/graph.py中构建图结构
节点与路由设置已完成，现在需要将所有元素连接起来。直接传递边和条件反馈边用于连接这些节点，最终生成的StateGraph能够运行整个工作流程。请将此代码放入src目录下的新文件graph.py中：
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()
构建过程是逐步进行的：
StateGraph(AgentTeamState)使用AgentTeamState模式创建一个有状态的图结构，这样随着工作流的推进，每个节点都能达到相同的共享状态。workflow.add_node(name, func)会将每个 Python 函数注册为节点。这些函数各司其职：负责设计测试、生成代码、运行测试或调试错误。workflow.set_entry_point("architect")确定了起始点。以初始状态调用或流式处理该图结构时，执行将从architect节点开始。
workflow.add_edge("architect", "developer") 和 workflow.add_edge("developer", "test_runner") 构成了最初的直线流程。当架构师完成需求规格与测试用例后，开发人员接手工作；而当开发人员编写出代码后，测试人员便会开始测试。workflow.add_edge("debugger", "test_runner") 则形成了自我修复循环，从而使普通的流程管道转变为具备自我修复功能的系统。一旦发现已修复的错误，流程不会继续传递给审核人员，而是将修复后的代码送回测试人员手中，以便再次验证修复效果。workflow.add_conditional_edges("test_runner", route_after_test_runner, ...) 用于在测试完成后让 LangGraph 调用 route_after_test_runner(state)。该函数的返回值会决定工作流是发送给 debugger 还是 reviewer。workflow.add_edge("reviewer", END) 用于结束运行流程。在审核者完成质量检查与总结后，图执行即告终止。workflow.compile() 会将图定义转换为可运行的图结构，同时验证其结构并为其执行做准备。第 8 步：src/main.py 中的 CLI 入口点
最后一部分是一个使用 rich 构建的命令行入口点，它通过带样式的终端面板实时显示每个子代理的进度。src 目录下的 main.py 文件以横幅和配置表显示函数开头：
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()
这些内容纯粹与终端样式相关，含义显而易见，因此无需进一步详解。同一个文件中接着是 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() 是各部分交汇的地方。它负责创建共享的 AgentTeamState，构建自修复图结构，并逐个节点流式执行任务。每个节点执行完成后，它会刷新当前状态，并在 Rich 面板中显示进度信息：架构师的规格与测试、开发者的代码、测试运行器的结果、每轮调试情况以及审核者的最终报告。最后会输出经过验证的实现代码、生成的 pytest 测试套件，以及总耗时、调试次数和测试结果等运行指标。
CLI 参数解析器用于完成文件的其余部分：
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."
终端输出显示以下流程：
- 第1阶段（共5阶段）：首席架构师分析速率限制器的相关需求，编写了一套多线程版的
pytest测试套件，用于检测时间戳裁剪、滑动窗口阈值控制以及多线程并发请求处理情况。 - 第2阶段（共5阶段）：开发人员基于
collections.deque和threading.Lock编写了第一个SlidingWindowRateLimiter类。 - 第3阶段（共5阶段）：在第一次迭代中，测试运行器执行
pytest测试。由于时间戳裁剪操作在加锁的关键代码段之外进行，导致了竞态条件，使得多个线程突破了请求限制，因此并发压力测试失败。
self._lock之前就调用了_prune_expired_timestamps()；解决方案是将时间戳清理操作放在同步代码块中执行。pytest，最终测试结果为“0.04秒内通过6项测试”，显示为绿色。本次运行时终端输出的截图：
完整输出还包括所编写的Python代码以及用于测试的用例。在此全部列出只会增加篇幅而并无太大价值。
该模式的适用场景
通过结合专门的子智能体、为每个子智能体设定不同的模型与思维层级，再加上确定性的执行沙箱，原本脆弱的单次提示工作流便能转变为更为可靠、具备自我修复能力的系统。
演示中使用的是代码生成团队，但同样的状态机模式也适用于许多其他工作流，比如数据管道、动态API编排以及结构化文档验证。核心理念很简单：不要依赖单一模型来完成所有任务。为每个智能体分配明确的职责，让系统在运行过程中自行检查并纠正错误。
有三点经验尤为突出：
- 绝不能让大语言模型自行评估自己的代码。应将执行任务交给运行
pytest的本地确定性子进程沙箱，这种方式速度快、无需额外成本，且能给出明确的通过或失败结果。