Самовідновлювана команда багатьох агентів у LangGraph: архітектор, тестувальник, дебагер
Створіть машину станів LangGraph на Python, у якій Архітектор, Розробник, сандбокс 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,
}
Кілька деталей роблять цю перевірку надійною без будь-якої думки ШІ:
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 та додавайте вузли по одному.
Допоміжні парсери для вилучення коду та розділення специфікацій
Два невеликі інструменти допомагають підтримувати порядок у процесі обробки. LLM часто оформлюють код у рамки 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
Кожен помічник вирішує певну поширену проблему у результаті роботи LLM:
_get_text()нормалізує формат відповіді. Залежно від постачальника та обгортки LangChain, відповідь може надходити у вигляді звичайного рядка або структурованого контенту, наприклад списку з елементамиtypeтаtext. Помічник перетворює обидва варіанти на звичайний рядок у Python._extract_code()витягує код із блоків Markdown. Він використовує регулярні вирази з параметромre.DOTALL, щоб зафіксувати багатокроковий код Python незалежно від того, чи було написаноpythonпісля відкриваючих тильд. Якщо блоку Markdown немає, він використовує очищений текст відповіді.
_parse_architect_output() розділяє відповідь Архітектора за допомогою роздільника ### TESTS. Усе, що знаходиться до цього маркера, стає specs, а все, що після — tests. Це дозволяє відокремити специфікацію у природній мові від коду тестів, який згодом передається виконувачу тестів.Вузол Lead Architect
Lead Architect є вхідною точкою команди, і він працює за принципом «спочатку тести». Ще до створення будь-якої реалізації він вивчає запит користувача, визначає структури даних та інтерфейси класів, а також розробляє ретельний набір тестів 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"}.
Вузол детермінованого тестового сандбоксу
Це один із найважливіших вузлів системи. Замість того, щоб запитувати ШІ, чи правильний код, він запускає набір тестів 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
Після проходження кожного тесту або при досягненні ліміту спроб керування переходить до вузла 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з рівнем мисленняlow, що підходить для швидкого створення резюме. На цьому етапі потрібно мало логічних міркувань, тому більш легка модель з нижчою здатністю до мислення забезпечує кращу швидкість та економію коштів. - Контекстуальний огляд. Особа, яка переглядає, отримує остаточну реалізацію, показники успіху та невдач з сандбоксу, кількість ітерацій дебаггингу та повну історію дебаггингу.
- Оновлення стану. Вузол повертає
{"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.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 та показники виконання, такі як загальний час, кількість ітерацій відлагоджування та результати тестів.
Парсер аргументів командного рядка завершує роботу з файлом:
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 та перевірка структурованих документів. Основна ідея проста: не покладайтеся на одну модель для виконання всього. Надайте кожному агенту чітку відповідальність та дозвольте системі перевіряти та виправляти свою роботу протягом її виконання.
Виділяються три уроки:
- Ніколи не дозволяйте ШІ оцінювати власний код. Передайте виконання локальній детерміністичній сандбокс-процесу, який використовує
pytest— це швидко, безкоштовно та дає точну відповідь про успіх чи невдачу.