Самааднаўляльная каманда агентаў у LangGraph: архітэктар, тэсціроўшчык, адладчык
Збіраем на Python машыну станаў LangGraph, дзе архітэктар, распрацоўшчык, пясочніца pytest, адладчык і рэцэнзент працуюць разам і паўтараюць цыкл, пакуль тэсты не пройдуць.
Крыніцы: Build a Self-Healing Multi-Agent Team with LangGraph
Штучныя інтэлекты ўжо выконваюць ролі практычна ў кожным слою працы з програмнамі, — ад стварэння коду да аналізу дадзейнаў і багатаэтапнага абдумвання. Аднак кожны, хто прасіць большую мовную мадэль аб чым-небудзь складным, сталквіваецца з тым самым проблемам. Адпаведзь на адной спробы можа здавацца правільной на першы погляд, але згодзя з выкананнем усіх вакалентаў і дзеякіх спецыяльных сітуацый яна перастае быць корысной. Звычны спосаб — ручная перапрацоўка: копіюецца лог адказаў, вставляецца ён у чат, прасіцца аб парадку, і так паказваецца, пакуль усё нарэшце не запрацюе.
Адваротны падход таксама мае сваія проблемы. Даведчы аднаму монолітнаму агенту дзесяткі інструментоў і дужы спіс інструкцый, часта падвышаецца рэглі ўпалоў, таму што яго кантэкст пачынае быць забрудненым, а характэрыстыкі інструментоў становяцца незрозумелымі. Якасць адпаведзяў падымаецца, таму што агент не можа выконваць усія тыя інструкцыйы адразу.
Дзеянне, паказана ніжэй, рашыяе оба прыблэмы за дапамою модульнай, іерархічнай машыны станоў калькольнікаў, напісанай на Python з аднойчынай бібліятэкай LangGraph. Прыклад працы — група з працэў па інжынерыі програмнага забезпечэння, якая сама сабе дапамагае і складаецца з пяці ролей: архітэктара, распрацоўшчыка, сандбокса Pytest, які работае на аднойчынных падпроцэсах, дэбагера і аднародніка. Гэты патэрн можа быць выкарыстоўваны ў багацькох іншых робочых процэсах інжынерыі ШЫ.
Разбіўка вялікай задачы між спецыялізаванымі падкалькольнікамі, выдача кожнаму належнага колічыні можлівасцяў для аналізу, адработка рэзультатаў у кантролюемым сандбоксе і наявнасць стабільнага цыклу зворотнай свярзкі дазволяюць групе тэставаць свою работу, выяўляць прыблэмы і іх ліквідаваць прычым перад тым, як перэдаць пераканальваны рэзультат. Спачатку — адзін взгляд на архітэктуру.
Сістэму формуюць два прынцыпы.
Першым ўскладненням являецца гетэрагенная іўерніця мадэляў з адаптывнымі рывнямі мышлення. Ролі відрызняюцца за складнісцю, таму не всі з яўляюцца однаковымі за мадэллю чы рэсурсамі обчыслення:
- «Архітэктор» працюе на
gemini-pro-latestз рывнем мышленняhigh. Ён віконуе найбольш складную частку працы, такую як встановлення меж проблемы, перэдбачэнне сітуацый конкурэнціі і стварэнне строгіх набораў тэстоўpytest. - «Распрацоўшчык» працюе на
gemini-3.7-flashз рывнем мышленняmedium. Цэе збалансавана настройка для шыранай сінтэзы коду, якая следуе часткавым вказівкам «Архітэктора». - «Запускчык тэстоў» — цэ гэта дэтэрміністычны падпроцес на Python, які запускае
pytestлокальна за прыблызна 15 мілісекунд і зовсім не выкарыстоўвае токены API LLM.
gemini-3.7-flash з рэжымам high для аналізу, таму што адстэпаванне па первасных заданнях да лініі з бягамі вымагае аналізу прычын-наследкаваў.gemini-flash-lite-latest з рэжымам low для аналізу. Перакананне ў стылі і напісанне падсумковага апісання для карыстальніка выконваецца за дапамогою простага разумавання.Другі прынцып — строгая абмежэнасць інструментаў, тое ўсе агенты працуюць з мінімальнымі правамі. Жадны падагент не мае необмежанага доступу да тэрміналу і сістэмы файлаў. Кожны агент отрымае толькі тыя даны і інструменты, якія неабходны для яго ролі. Распрацоўшчык не можа запускаць будзь-якія команды шэлу, а агент для тэстав проста выконвае тэсты і апісвае тое, што на самай працэ выйшло, не прасячы ШЫ аб адгукве чы інтэрпрэтэцыі рэзультатаў. Строгая абмежэнасць дапамагае кожнаму агенту заставацца сфокусаваным і зменшае можлівасць галюцынацый.
Канцэпцыі LangGraph, якія выкарыстоўваюцца
LangGraph працюе як спяльная дошка. Кожны член команды, тыямо кажучы, кожны падагент, выканае сваю задачу і запісвае рэзультат у той самы спяльны простор, так што всі бачаюць кантэкст, які каманда створыла да гэтага часу. LangGraph яўляецца оркестрактаром: ён вяршыць рашэнне, які падагент будзе працаваць далей, і перадае відповідны стан.
Тут важлівы чатыры ідэі:
- Стан, які ўособліваеся за дапамою
TypedDict, являецца спяльной дошкай. Усі падагенты чытаюць і запісваюць у аднам Python-слоўніку пад назвайAgentTeamState, які мае ў сабе запіт корыстніка, спэцыфікацію, тэсты на елементарнасць, чырвоныя версіі коду, рэзультаты тэстаў і колькасць раундаў дэбагавання. - Вузлы — гэта члены каманды. Вузел — гэта проста Python-функцыя, якая прыме настоячы стан, выканае сваю задачу (звярнуцца да LLM, запрацаваць тэсты) і вернуе абнаўленні, якія злучаюцца звярнуць у спяльны стан.
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, таму модэлі на кшталт GPT 5.6 чыста Deepseek v4 Pro можна заменіць, калі не хочацца апыляцца толькі на Gemini.pytest— гэта адпаведальны за тэстыранне ўнутры локальнага падпроцэсу. Ён запускае створаны набір тэстаў і вяртае рэальныя рынкі.richфарматуе выходны тэкст у тэрмінале за дапамогою калераў, табэляў і выявлення сінтаксісу, што спрыяе лёгкаму адстэпленню за працай়бой агента і яе дыягностыцы.python-dotenvаўтаматычна чытае клучы API чыста настройкі модэляў з файлу.env.
Файл .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
Калі сераўнік готавы, аплікацыю можна ствараць пашагоўна.
Шаг 1: спяльны стан у src/state.py
Перш за ўсё неабходна з’явіць схему дадзеных, якія пераходзяць межы агентаў. Створыце каталог src з файлам state.py унутры яго.
Першыя ваказанні — гэта мадэль TestResult, напісаная за дапамогою Pydantic. Яна прэдставляе структураваны рэзультат запуску 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, якое містыць толькі значныя часткі апісу працэсу, якіе дыбаггер выкарыстоўвае для знаходжэння і выправлення прычыны неудачы.
Далей ідзе главны элемент AgentTeamState у фармаце TypedDict. Цэтае ўособленне служыць «дошкай для заметак», якая пераводзіцца па графе: кожны вузел чытае тое, што яму трэба, выкананае ўсё неабходнае і апдэйтуе поля, на якія паслялеглі вузелы залежаць.
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 прызначае правы модэль і рывень разумовых працэсаў (low, medium або high) кожнай ролі пад-агента. Файл 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, тады можна налаштаваць ступеньню аналізу па кожной ролі. А кожны кліент, які ён вяртае, рэалізуе BaseChatModel з LangChain, таму вузлы графа вызываюць .invoke() адносова, незалежна ад таго, які прадастык знаходзіцца за роллю.
Шаг 3: дэтэрміністычны сандбокс pytest у src/sandbox.py
Частая памылка ў дызайне агента — прасіць LLM пераканацца, у своем “разуме”, у правільнасці коду, які ён толькі напісаў. Мадэль можа з упэўненасцю прыняць, што код праходзіць пераканання, хоць у яме ёсць багі, таму што мадэлі часта выдумваюць, што ўсе іхні рэзультаты правільныя. Нават калі мадэль і выявляе дзеякі багі, нема гаранцыі, што ёй вдалося знайсці ўсе.
Болей надзяйны спосаб — дазволіць коду самаму сабе падтвердзіць свою эфектывнасць. У зменшэнні на тое, каб паверыць у рашэнне модэлю, локальны вузел выканання запісвае створаны код і тэсты ў тымчасовы каталог і запускае pytest у ізольаваным падпроцэсе Python. Рэзультатам являецца об’ектыўнае пасвярджэнне аб аббыцці, якое можа буць безпосерэдна включанае ў працэс працы.
Створыце sandbox.py у папцы src. Ён пачынаецца з дапаможнага функцыі _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,
}
Кальколька дэталей робяць гэты аптрымненне надзеяным без якога-лібо мнення LLM:
tempfile.TemporaryDirectory()задае кожнам запуску чыстую рабочую прастору. Там пісваецца ствароўваны код і тэсты, а калі завершаецца выкананне, дырэктарый лічыцца.timeout=15лімітуе час выканання. Код, які застраг у бесконечным цыкле або працуе занадта давго, мае свой падпроцес убівацца і не можа блакануць весь працэс агента.exit_code == 0— гэта дэтэрміністычны сігнал пра успех або неудачу адpytest. Ноль означае, што тэставанне прыйшлося, а будзь-які іншы код — проблему; таму рэзультат і стэк-трэйс падаюць у дэбаггер.
error_summary. Гэтая структура знаходзіцца ў спадзеланым стане, і граф выкарыстоўвае яе, каб адначасова прыняць рашэнне пра продаж чы вход у цыкл самавылечэння.Шаг 4: запіты сістэмы для кожнага спеціяліста ў src/prompts.py
У мнага-агентный сістэме запіты должны быць короткімі, чыткімі і мець часткава адзначаны дыячны спектар. Кожнам агенту патрэбна чыстая картына яго задач, обавязакоў і меры, без непатрэбных інструкцый і без перакрыцча з агентамі-суседзямі. Файл prompts.py у падкаталогу src візначае чатыры разныя сістэмныя запіты для спецыялізаваных ролей:
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, віконуе певну задачу, напрыклад, віклікае LLM або запускае тэсты, і повертае частыні, які потрэбуюць оновлення. LangGraph об’еднуе іх у спільній стан, тож наступній вузок завжды мае найновішый кантэкст.
Створьце файл nodes.py у папці src і додайце вузкі по одному.
Допомогові засобі для віцягування коду і падзелу спецыфікацій
Дві маленькі утыліты допомагаюць пітрымуваты чыстоту у працэсі. LLMs часта обгортаюць код у 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
Każы падпамогчык адрабатвае загальныя проблемы ў выходных даных LLM:
_get_text()нормалізуе фармат адпаведzi. Залежна ад прадаўцы і кэшувальніка LangChain, адпаведзь можа прыйсці як звычайны рэчыскап або як структурованы кантэнт, напрыклад, у відзеце зtypeіtext. Падпамогчык ператварае яго ў звычайны рэчыскап на Python._extract_code()выкарыстоўвае рэгулярныя выразы з параметрамre.DOTALL, каб выкарыстаць код з блакоў Markdown. Якім бы не быў фармат, 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")насталяваеgemini-pro-latestз рэжымам разважоўкіhigh. У гэтым случае корыстна сільная мадэль з глęбокім аналізам, таму што трэба спланаваць усё на адной стадыі, включаючы розгляд крайніх сцэнарыяў, такіх як безпека потакаў, порожнія колекцыі чы абмежэння. - Стварэнне паведамлення. Сістэмны прыказ
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 і тэставыя асэрцыі, таму разрабоўцу не трэба нічога перадзяляваць, і ён можа сфокусавацца на шырокім і точным кодзе. Для гэтага дастатнека срэдніх зусілляў, а таксама гэта зменшывае выкарыстанне токенав і вартасць разработкі. - Аднесенне да набора тэстаў.
state['tests']прымешчаецца безпасова ў запрос разрабоўцу. Бачыць точныя асэрцыі, якія трэба пераверыць, дапамагае прыемніку прыкрэпіцца да апэктыўных імен методаў, тыпаў аргументаў і значэнняў, якія яны маюць вернуць. - Абнаўленні стану. Пасля выкарыстоўвання чыстага Python у адпаведзі, вузел вяртае
{"code": code, "status": "code_generated"}.
Вузел дэтэрміністычнай песочніцы тэстаў
Эта ўзловая частка являецца аднай з найважлівейшых у сістэме. У зяместо таго, каб запытаць LLM, чы рэштры выглядаюць правільна, праеўначваецца набор pytest на код у ізольаваным падпроцесе. Тэсты выконваюцца локальна, таму гэты крок не выканае жадных вызоў API LLM і зазвычай завершаецца за апоўну 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
Пасля таго, як кожны тэст праходзіць, або калі досягнуты ліміт пракатоў, кантроль пераходзіць да вузла Reviewer – фінальнага контроля якосці. Ён пераканальваецца ў загальной якосці коду, пераканальваецца, што докстрынгі ўсё ясныя і корыстныя, і пішае кароткі апূвяданні пра завершаны рашэння для карыстальніка. Функцыя 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")вяртаеgemini-flash-lite-latestзнізкімрывень разумавання, якая падходзіць для шыранага стварэння апূরкі. На гэтым этапе не патрэбна значная колькасць разумавання, таму лёгкія модэлі з меншымі можлівасцямі разумавання даюць вышэю шыроцу і ніжэйшыя вітраты. - Аналіз у кантэксце. Адміністратар апূркі отрымае гэтую фінальную рэалізацыю, показнікі успеху і нявераго з сандбокса, колькасць ітэрацый дыбагування і цэлую історыю дыбагування.
- Абнаўленні стану. Вузел вяртае
{"review_summary": _get_text(response.content), "status": "completed"}.
Шаг 6: логіка маршрутацыі ў файле src/edges.py
Канальная логіка і вузлы агентаў ўжо створэны, таму наступны крок — маршрутызацыя. Вузлы выкананяюць обчысленні, а рэбра вялікія, куда пайдзе кантроль. Умовнае рэбра дзейніцца як дынамічны маршрутызатор: ён аналізуе словнік стана, які вярнуў вузел, і дае назву наступнага вузла для выканання.
Функцыя маршрутызацыі знаходзіцца ў новам файле edges.py унутрь 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"
Раутэр аналізуе рэзультаты тэстоў і колькасць прабаў дыбаггавання, і ён развязвае тры сцэны:
- Успешны шлях (
passed == True). Калі всі тэсты праходзяць за першай прабыткі, працэс праходзіць адразу да Рэвізора, і няма прычыны вітрачаць дадатковы час або рэсурсы на Дыбаггер. - Самаўздоровленне (
passed == Falseкаліiteration < max_iterations). Тэсты, якія не праходзяць, але ўсё шчэ ёсць прабыткі, адправляюць код да Дыбаггера, які аналізуе прычыну неудачы і прабуе яе выправіць. - Дасягнуты ліміт прабаў (
iteration >= max_iterations). Калі Дыбаггер выкарыстаў усе тры дазволеныя прабыткі, а тэсты ўсё шчэ не праходзяць, працэс праходзіць да Рэвізора, а не застаецца у бесканечным цыкле. Гэты жорсткі ліміт самаўздоровлення запобегае застою працэсу з-за суперсучыяных аб нечыста сформульаваных трэбаванняў.
Шаг 7: стварэнне графа ў src/graph.py
Вузлы і маршрутызацыя вжо створэны, таму цяпер усё паў’язваецца. Прамыя зв’язкі і умовныя канцэры з’еднаюць вузлы, і рэзультат компілюецца ў StateGraph, які можа запускаць усю робочую практыку. Паместыце гэты код у новы файл graph.py ўнутрь папакі 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()
Стварэнне адбываецца пасля пасля:
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 да пераглядача. Пасправлены код вяртаецца да тэставальніка, ўпэўніваючыся, што пасправка зноў перавернута.workflow.add_conditional_edges("test_runner", route_after_test_runner, ...) выклікае у LangGraph функцыю route_after_test_runner(state) пасля таго, як завершаюцца тэсты. Значэнне, якое яна вяртае, направляе рабочы план або да debugger, або да reviewer.workflow.add_edge("reviewer", END) завершае експлуатацыю. Пасля таго, як reviewer завершыў перакантроль якосці і стварэнне падсумку, выкананне графа заканчываецца.workflow.compile() ператварае апіс графа ў граф, які можна запусціць, перакантролюючы яго структуру і падготавляючы яго да выканання.Шаг 8: тэчка входу CLI у файле src/main.py
Пасёлкі ўсьмо — гэта тачка запуску з лініі каманд, створаная за дапамою rich; яна у рэальны час прадае інформацыю пра прагрэс кожнага пад-агента за дапамою стайляваных панелей тэрміналу. Файл main.py у падкаталозе src пачынаецца функціямі для адразування заголовка і прадастоўвання табліцы налашчэнняй:
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 этапаў распрацоўшчык піша першы клас
SlidingWindowRateLimiter, пабудаваны наcollections.dequeіthreading.Lock. - На 3-м з 5 этапаў працоўнік тэставання на ітэрацыі 0 запускае
pytest. Тэст на навантажэння з адночаснымі запыткамі не выйшаў, таму што адсечкі часовых пазначэнняў выканаліся за межамі заблокаванага крытычнага сектара, і ўнаследак супернагонкі калькі праменяў перасілі дазволеныя ліміты запыткаў.
_prune_expired_timestamps() была выклікана раней, чым быў захаваны self._lock; таксама апрантлер практыкуе выкарыстоўванне сінхронізаванага блока для адбыцься выдалення таймстампаў.pytest на падправленым кодзе, і рэзультат тэстаў паказваеся як «зелены», з напісам 6 прынятыя за 0.04 секунд.Скрыншот рэальнага выходу з тэрміналу пасля гэтага запуску:
Полны выход таксама уключае код на Python, які быў напісан, і тэстовыя кейсы, якія вікорыстоўваліся для ўтверджэння яго. Перадрукаванне усяго гэтага тут прынесла б дадатковую дужыну без значныя корысты.
Дзе застосоўваецца гэты патэрн
Саюз спецыялізаваных падагентоў, разных модэляў і рубежаў мышлення для кожнага з іх, а таксама сандбокса для дэтэрміністычнай експансіі ператварае кранглавы працэс на адной прампт-запросе ў болей надзеяны, самовылечваючыся сістэму.
Дэманстрацыяю была команда для генеравання коду, але той самы шаблон машыны станоў падходзіць для багацых іншых працэсоў, такіх як канвеі дадзеных, дынамічная оркестрацыя API і верыфікацыя структураваных дакументаў. Основная ідэя простая: не павінавацца на адны модэль для выкаання всего. Дай кожнаму агенту чыстае заведанне і дазволь сістэме перакантроліваць і пасправляць свою работу па ходу.
Выдзеляюцца тры урокі:
- Ніколы не давайце LLM ацэніваць свой сабе код. Перадайце выкаання локальнаму дэтэрміністычнаму падпрацэсу-сандбоксу, які запускае
pytest— ён шырок, бесплатны і дае точную адказу пра успех чы неудачу.