Judge Panel Tournament Pattern для параллельной генерации и ранжирования MVP-вариантов
Обзор паттерна
Паттерн "параллельные варианты + турнир/judge panel" — это архитектурный подход для одновременной генерации N (3-10) различных вариантов решения одной задачи и автоматического ранжирования их через систему судей (judge agents). Этот паттерн решает три ключевые проблемы:
- Параллелизм: изоляция контекста каждого варианта (отдельные sandbox, git worktree, контейнеры)
- Оценка: объективное ранжирование вариантов через разные стратегии (knockout, round-robin, single-pass)
- Объяснимость: визуализация результатов для демонстрации клиенту
1. Архитектура параллельного выполнения N агентов
Уровень контекста и изоляции
Git Worktree подход (для кодогенерации)
project/
├── .git/ (shared)
├── worktree_variant_1/
│ ├── src/
│ ├── tests/
│ └── .branch (git worktree link)
├── worktree_variant_2/
│ ├── src/
│ ├── tests/
│ └── .branch (git worktree link)
└── worktree_variant_3/
├── src/
├── tests/
└── .branch (git worktree link)
Команда инициализации (LangChain/CrewAI compatible):
import subprocess
import concurrent.futures
def init_worktree(variant_id: str, base_branch: str):
branch_name = f"variant-{variant_id}"
subprocess.run([
"git", "worktree", "add",
f"worktree_variant_{variant_id}",
base_branch
])
subprocess.run([
"git", "-C", f"worktree_variant_{variant_id}",
"switch", "-c", branch_name
])
return f"worktree_variant_{variant_id}"
# Параллельное создание 5 worktree для 5 MVP-вариантов
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
worktrees = list(executor.map(
lambda i: init_worktree(str(i), "main"),
range(5)
))
Docker контейнер подход (для сервис/backend-генерации)
# docker-compose.yml для 5 параллельных вариантов
services:
variant-1:
build: .
volumes:
- ./variant_1:/workspace
environment:
VARIANT_ID: "1"
AGENT_SEED: "42001"
variant-2:
build: .
volumes:
- ./variant_2:/workspace
environment:
VARIANT_ID: "2"
AGENT_SEED: "42002"
# ... повтор для variant-3..5
In-Memory Sandbox подход (для быстрой итерации)
from dataclasses import dataclass
from copy import deepcopy
import asyncio
@dataclass
class MVPVariant:
variant_id: str
context: dict # изолированный контекст
outputs: dict # результаты выполнения агента
async def run_variant_agent(variant: MVPVariant, agent_fn, task_prompt):
"""Запуск агента в изолированном контексте"""
# Копируем глобальное состояние для этого варианта
local_context = deepcopy(variant.context)
# Агент работает только с local_context
result = await agent_fn(
prompt=task_prompt,
context=local_context,
variant_id=variant.variant_id
)
variant.outputs = result
return variant
# Параллельное выполнение 5 агентов одновременно
variants = [
MVPVariant(f"mvp_{i}", {}, {})
for i in range(5)
]
tasks = [
run_variant_agent(v, my_agent_function, task_prompt)
for v in variants
]
completed_variants = await asyncio.gather(*tasks)
Управление состоянием и зависимостями
- Per-variant environment: каждый вариант имеет собственный
VARIANT_ID, seed для LLM reproducibility - Shared resources: лишь read-only доступ к общей knowledge base, документации
- Context isolation: переменные, history, embeddings кешируются отдельно для каждого варианта
Рекомендация для LangChain:
from langchain.globals import set_debug
from langchain.callbacks import FileBasedMessageHistory
# Каждый вариант → отдельный callback handler
handlers = {
f"variant_{i}": FileBasedMessageHistory(
file_path=f"logs/variant_{i}.json"
)
for i in range(5)
}
2. Judge/Critic агент: архитектура, критерии, rubric
Структура Judge агента
Judge agent — это специализированный LLM, который оценивает выходы других агентов по чётким критериям:
from enum import Enum
from pydantic import BaseModel
class EvaluationCriterion(BaseModel):
name: str # "code_quality", "performance", "maintainability"
weight: float # 0.0 - 1.0
rubric: dict # уровни оценки и описания
evaluator_prompt: str # специфичный промпт для judge
class JudgeConfig:
criteria: list[EvaluationCriterion] = [
EvaluationCriterion(
name="functional_correctness",
weight=0.4,
rubric={
"excellent": "Полностью соответствует spec, все edge cases покрыты",
"good": "Работает для основных сценариев, 1-2 edge case пропущено",
"acceptable": "Работает для основного пути, есть баги",
"poor": "Не работает или критические баги"
},
evaluator_prompt="Проверьте следует ли код спецификации..."
),
EvaluationCriterion(
name="code_quality",
weight=0.3,
rubric={
"excellent": "Clean code, хорошие имена, документация",
"good": "Читаемый код с минимальной документацией",
"acceptable": "Работает но сложен для понимания",
"poor": "Грязный код, без структуры"
},
evaluator_prompt="Оцените качество кода..."
),
EvaluationCriterion(
name="performance",
weight=0.2,
rubric={
"excellent": "O(n) или better, умное использование структур",
"good": "Разумная сложность, без очевидных оптимизаций",
"acceptable": "Работает но медленнее оптимального",
"poor": "Серьёзные performance проблемы"
},
evaluator_prompt="Проанализируйте сложность..."
),
EvaluationCriterion(
name="maintainability",
weight=0.1,
rubric={...}
)
]
Scoring система
class JudgeEvaluation(BaseModel):
variant_id: str
criterion_scores: dict[str, float] # "code_quality" -> 0.85
weighted_score: float # sum(score * weight)
reasoning: str # объяснение для демо клиенту
strengths: list[str]
weaknesses: list[str]
class JudgeAgent:
def evaluate_variant(
self,
variant_output: str,
variant_id: str,
config: JudgeConfig
) -> JudgeEvaluation:
"""
Используя config.criteria, оцениваем вариант по каждому критерию.
Итоговый score = sum(score_i * weight_i)
"""
scores = {}
reasoning_parts = []
for criterion in config.criteria:
# Вызываем LLM с специфичным промптом
llm_evaluation = self.llm.invoke(
f"{criterion.evaluator_prompt}\n\nВариант:\n{variant_output}"
)
# Парсим ответ LLM в числовую оценку (0.0-1.0)
score = self._parse_rubric_response(
llm_evaluation,
criterion.rubric
)
scores[criterion.name] = score
reasoning_parts.append(llm_evaluation)
# Взвешенный итоговый score
weighted = sum(
scores[c.name] * c.weight
for c in config.criteria
)
return JudgeEvaluation(
variant_id=variant_id,
criterion_scores=scores,
weighted_score=weighted,
reasoning="\n".join(reasoning_parts),
strengths=self._extract_strengths(reasoning_parts),
weaknesses=self._extract_weaknesses(reasoning_parts)
)
3. Стратегии ранжирования: Knockout vs Round-Robin vs Single-Pass
Стратегия 1: Single-Pass Ranking (экономичный подход)
Каждый Judge оценивает все варианты один раз.
Плюсы: минимум LLM calls (N вариантов × M judges) Минусы: нет перепроверок, может пропустить нюансы
async def single_pass_ranking(
variants: list[str],
judge_model: str = "claude-3-5-sonnet-20241022"
) -> list[JudgeEvaluation]:
"""Один проход: каждый judge оценивает все варианты один раз"""
judge = JudgeAgent(model=judge_model)
config = JudgeConfig()
evaluations = []
for i, variant in enumerate(variants):
eval_result = judge.evaluate_variant(
variant_output=variant,
variant_id=f"variant_{i}",
config=config
)
evaluations.append(eval_result)
# Сортируем по weighted_score
return sorted(
evaluations,
key=lambda e: e.weighted_score,
reverse=True
)
# Cost: 5 вариантов × 1 judge = 5 LLM calls
Стратегия 2: Round-Robin Tournament (точность выше)
Варианты "играют" друг с другом, judge сравнивает пары.
Плюсы: относительная оценка часто точнее абсолютной, меньше subjectivity Минусы: больше LLM calls (~N² / 2), slower
from itertools import combinations
async def round_robin_tournament(
variants: list[str],
judge_model: str = "claude-3-5-sonnet-20241022"
) -> list[tuple[str, float]]:
"""Round-robin: каждый вариант vs каждый другой"""
judge = JudgeAgent(model=judge_model)
# Счёт побед для каждого варианта
scores = {f"variant_{i}": 0 for i in range(len(variants))}
# Все пары сравниваем
for i, j in combinations(range(len(variants)), 2):
winner = judge.compare_pair(
variant_a=variants[i],
variant_b=variants[j],
criteria=JudgeConfig().criteria
)
scores[f"variant_{winner}"] += 1
# Сортируем по кол-ву побед
ranking = sorted(
scores.items(),
key=lambda x: x[1],
reverse=True
)
return ranking
# Cost: C(5,2) = 10 comparisons = 10 LLM calls
# Но для 10 вариантов: C(10,2) = 45 calls (дорого)
Стратегия 3: Knockout Tournament (быстро)
Вариант-победитель проходит дальше, проигравший выбывает.
Плюсы: логарифмическое кол-во LLM calls (N → log₂(N)), театральный effect Минусы: может быть unfair (bad bracket), не даёт полного ranking
import random
async def knockout_tournament(
variants: list[str],
judge_model: str = "claude-3-5-sonnet-20241022"
) -> dict:
"""Knockout tournament: single-elimination bracket"""
judge = JudgeAgent(model=judge_model)
# Инициализируем bracket с random order
bracket = variants.copy()
random.shuffle(bracket)
round_num = 0
while len(bracket) > 1:
round_num += 1
next_round = []
# Парные поединки
for i in range(0, len(bracket), 2):
if i + 1 < len(bracket):
winner = judge.compare_pair(
variant_a=bracket[i],
variant_b=bracket[i+1],
criteria=JudgeConfig().criteria
)
next_round.append(winner)
else:
# Bye (чётное число вариантов)
next_round.append(bracket[i])
bracket = next_round
print(f"Round {round_num}: {len(bracket)} variant(s) remain")
return {
"winner": bracket[0],
"tournament_rounds": round_num,
"total_comparisons": len(variants) - 1
}
# Cost: для 5 вариантов = 4 comparisons; для 10 вариантов = 9 comparisons
# Экономичнее than round-robin, но меньше информации
4. Multi-Judge Panel (повышение надёжности)
Вместо одного judge, используем панель из 2-3 судей разных моделей:
class JudgePanelEvaluation(BaseModel):
variant_id: str
individual_scores: dict[str, float] # judge_name -> score
consensus_score: float # average или weighted median
panel_reasoning: str # согласие/несогласие судей
async def panel_ranking(
variants: list[str],
judge_models: list[str] = [
"claude-3-5-sonnet-20241022",
"gpt-4-turbo",
"gemini-pro"
],
consensus_method: str = "median" # median более robust чем mean
) -> list[JudgePanelEvaluation]:
"""Несколько судей оценивают каждый вариант"""
evaluations_by_variant = {}
for i, variant in enumerate(variants):
variant_id = f"variant_{i}"
individual_scores = {}
# Каждый judge оценивает независимо
tasks = []
for judge_model in judge_models:
judge = JudgeAgent(model=judge_model)
task = judge.evaluate_variant(
variant,
variant_id,
JudgeConfig()
)
tasks.append((judge_model, task))
results = await asyncio.gather(*[t[1] for t in tasks])
# Собираем оценки от всех судей
for (judge_model, _), result in zip(tasks, results):
individual_scores[judge_model] = result.weighted_score
# Консенсус: медиана более robust чем mean
scores_list = list(individual_scores.values())
if consensus_method == "median":
consensus = sorted(scores_list)[len(scores_list)//2]
else: # mean
consensus = sum(scores_list) / len(scores_list)
evaluations_by_variant[variant_id] = JudgePanelEvaluation(
variant_id=variant_id,
individual_scores=individual_scores,
consensus_score=consensus,
panel_reasoning=f"Судьи согласны? {max(scores_list) - min(scores_list) < 0.2}"
)
# Финальный ranking по consensus_score
return sorted(
evaluations_by_variant.values(),
key=lambda e: e.consensus_score,
reverse=True
)
# Cost: 5 вариантов × 3 судей = 15 LLM calls
5. Визуализация результатов для демо клиенту
Структурированный отчёт
from dataclasses import asdict
import json
class ComparisonReport:
def __init__(self, evaluations: list[JudgeEvaluation]):
self.evaluations = evaluations
self.ranking = sorted(
evaluations,
key=lambda e: e.weighted_score,
reverse=True
)
def generate_html_report(self) -> str:
"""HTML отчёт с рейтингом, графиками, аргументами"""
html = """
<html>
<head>
<title>MVP Variant Comparison</title>
<script src="https://cdn.plot.ly/plotly-latest.min.js"></script>
<style>
body { font-family: -apple-system, sans-serif; margin: 20px; }
.variant-card {
border: 1px solid #ddd; padding: 15px; margin: 10px 0;
border-radius: 8px; background: #f9f9f9;
}
.rank-badge { font-size: 24px; font-weight: bold; color: #007bff; }
.score-bar {
width: 100%; height: 20px; background: #e0e0e0;
border-radius: 4px; overflow: hidden;
}
.score-fill { height: 100%; background: #28a745; }
.criteria-grid {
display: grid; grid-template-columns: repeat(4, 1fr); gap: 10px;
margin-top: 10px;
}
.criterion-box {
background: #fff; border: 1px solid #eee;
padding: 10px; border-radius: 4px; text-align: center;
}
</style>
</head>
<body>
<h1>MVP Variant Comparison Report</h1>
"""
# Ranking таблица
html += "<h2>Overall Ranking</h2><table border=1><tr><th>Rank</th><th>Variant</th><th>Score</th><th>Reasoning</th></tr>"
for rank, eval in enumerate(self.ranking, 1):
html += f"""
<tr>
<td><strong>{rank}</strong></td>
<td>{eval.variant_id}</td>
<td><strong>{eval.weighted_score:.2f}</strong></td>
<td>{eval.reasoning[:200]}...</td>
</tr>
"""
html += "</table>"
# Детальные карточки для каждого варианта
html += "<h2>Detailed Evaluation</h2>"
for rank, eval in enumerate(self.ranking, 1):
html += f"""
<div class="variant-card">
<div style="display: flex; align-items: center; gap: 20px;">
<span class="rank-badge">#{rank}</span>
<div style="flex: 1;">
<h3>{eval.variant_id}</h3>
<div class="score-bar">
<div class="score-fill" style="width: {eval.weighted_score*100}%"></div>
</div>
<strong>{eval.weighted_score:.2f}/1.00</strong>
</div>
</div>
<div class="criteria-grid">
"""
for criterion, score in eval.criterion_scores.items():
html += f"""
<div class="criterion-box">
<small>{criterion}</small>
<div style="font-size: 18px; font-weight: bold; color: #007bff;">
{score:.2f}
</div>
</div>
"""
html += """
</div>
<h4>Strengths</h4>
<ul>
"""
for strength in eval.strengths:
html += f"<li>{strength}</li>"
html += """
</ul>
<h4>Weaknesses</h4>
<ul>
"""
for weakness in eval.weaknesses:
html += f"<li>{weakness}</li>"
html += """
</ul>
<h4>Full Reasoning</h4>
<pre style="background: #f5f5f5; padding: 10px; overflow-x: auto;">
"""
html += eval.reasoning
html += """
</pre>
</div>
"""
html += """
</body>
</html>
"""
return html
def save_report(self, filepath: str):
with open(filepath, "w") as f:
f.write(self.generate_html_report())
# Использование
report = ComparisonReport(evaluations)
report.save_report("comparison_report.html")
# Клиент открывает в браузере, видит рейтинг + аргументы
6. Open-Source реализации на GitHub
1. LangGraph (LangChain)
- Репо: https://github.com/langchain-ai/langgraph
- Модель: DAG-based workflow, parallel branches для N агентов
- Judge pattern: используется
create_react_agent()+ custom evaluator node - Пример для tournament: https://github.com/langchain-ai/langgraph/tree/main/examples (search "competition", "tournament")
2. CrewAI
- Репо: https://github.com/joaomdmoura/crewai
- Встроенная поддержка parallel tasks через
concurrent_execution: True - Judge pattern: можно настроить через custom Tool + Agent role="Judge"
- Пример:
from crewai import Agent, Task, Crew
judge_agent = Agent(
role="Judge",
goal="Evaluate and rank MVP variants by quality criteria",
backstory="Expert evaluator with deep technical knowledge"
)
# Tasks для разных вариантов в параллель
tasks = [
Task(
description=f"Evaluate variant_{i}: {variant_code}",
agent=variant_agents[i],
expected_output="Working MVP code",
)
for i in range(5)
]
judge_task = Task(
description="Rank all variants and provide comparison",
agent=judge_agent,
expected_output="Ranked list with reasoning",
)
crew = Crew(
agents=variant_agents + [judge_agent],
tasks=tasks + [judge_task],
process=Process.hierarchical,
manager_agent=judge_agent,
verbose=True
)
3. AutoGen (Microsoft)
- Репо: https://github.com/microsoft/autogen
- Multi-agent groupchat для evaluation
- Judge pattern: через
AssistantAgent+UserProxyAgentв groupchat - Поддерживает функции для оценки выходов (
ConversableAgent.register_reply())
4. Anthropic Cookbook (best-of-n sampling)
- Репо: https://github.com/anthropic-ai/anthropic-sdk-python
- Технология: в каждом вызове Claude генерирует N вариантов, возвращает all для ranking
- Judge pattern: можно использовать другую Claude модель для evaluation
# Генерируем N вариантов в одном запросе
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=10000,
system="Generate 5 different approaches to this problem",
messages=[{"role": "user", "content": "Design an MVP for X"}]
)
# Потом используем judging для ranking
5. Swarm (OpenAI)
- Репо: https://github.com/openai/swarm
- Легковесный фреймворк для agent coordination
- Judge pattern: встроенный через
Coordinator+ routing functions
7. Рекомендуемая архитектура: Hybrid подход
┌─────────────────────────────────────────────────┐
│ Generate N MVP Variants (Parallel) │
│ - Agent 1 in worktree_1 (isolated context) │
│ - Agent 2 in worktree_2 (isolated context) │
│ - Agent 3 in container_3 (Docker sandbox) │
│ - ... │
└────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ Single-Pass Ranking (First Filter) │
│ - Single judge (claude-3-5-sonnet) evaluates │
│ - Quick filtering, O(N) calls │
│ - Top 3 variants advance │
└────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ Round-Robin on Top-3 (Higher Precision) │
│ - Panel of judges: Claude + GPT-4 + Gemini │
│ - Consensus scoring (median aggregate) │
│ - 3 comparisons × 3 judges = 9 calls total │
└────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ Generate HTML Report for Client Demo │
│ - Ranking table + detailed scorecards │
│ - Strengths/weaknesses per variant │
│ - Judge reasoning (объяснимость!) │
└─────────────────────────────────────────────────┘
Расчёт стоимости для 5 MVP-вариантов: - Генерация: 5 параллельных агентов (зависит от complexity) - Single-pass: 5 calls - Round-robin (top-3): 3×2×3 = 18 calls (панель из 3 судей) - Total: ~23 LLM calls (экономичнее pure round-robin для всех)
8. Ключевые метрики и best practices
| Метрика | Рекомендация | Причина |
|---|---|---|
| Кол-во судей | 2-3 | Median more robust than single judge |
| Кол-во критериев | 4-6 | Balance specificity vs evaluation time |
| Фильтр перед tournament | Top 50% | Экономит compute, убирает очевидный мусор |
| Seed для reproducibility | Per-variant | Гарантирует разнообразие без chaos |
| Parallel execution | AsyncIO/ThreadPool | 3-5x ускорение vs sequential |
Заключение
Паттерн "параллельные варианты + judge panel tournament" критичен для: - Качества: несколько подходов + объективная оценка > один агент - Скорости: параллелизм уменьшает wall-clock time - Объяснимости: структурированное ранжирование даёт аргументы для клиента - Масштабируемости: гибкий выбор ranking стратегии в зависимости от бюджета
Гибридный подход (single-pass filter → round-robin на top-N) даёт оптимальный баланс между cost и accuracy.
Ссылки и источники
- LangGraph: https://github.com/langchain-ai/langgraph
-
Официальная документация: https://langchain-ai.github.io/langgraph/
-
CrewAI: https://github.com/joaomdmoura/crewai
-
Документация parallel tasks: https://docs.crewai.com/
-
Anthropic best-of-n sampling: https://github.com/anthropic-ai/anthropic-sdk-python
-
Research paper: https://arxiv.org/abs/2209.14375 (Constitutional AI, использует best-of-n)
-
Microsoft AutoGen: https://github.com/microsoft/autogen
-
Paper: https://arxiv.org/abs/2308.08155 (AutoGen: Enabling Next-Gen LLM Applications)
-
OpenAI Swarm: https://github.com/openai/swarm
-
Примеры multi-agent coordination
-
Judge/Evaluator patterns в LLM:
- https://arxiv.org/abs/2306.05685 (Judging LLM-as-a-Judge)
-
https://arxiv.org/abs/2310.03629 (Evaluating Large Language Models as Judges)
-
Git worktree для isolation: https://git-scm.com/docs/git-worktree
-
LangChain agent parallelization: https://python.langchain.com/docs/modules/agents/agent_types/openai_functions_agent
-
Scoring rubrics for LLM evaluation: https://github.com/langchain-ai/langsmith/tree/main/python/examples
-
Tournament algorithms: https://github.com/topics/tournament-algorithm
- Round-robin: https://en.wikipedia.org/wiki/Round-robin_tournament
- Knockout: https://en.wikipedia.org/wiki/Single-elimination_tournament