04 - 다중 에이전트 코딩: LangGraph, CrewAI 및 AutoGen
복잡한 작업에 직면해 있다고 상상해 보십시오: 처음부터 인증 시스템을 구현하는 것입니다. OAuth2, RBAC, 감사 로그, 통합 테스트가 포함된 엔터프라이즈 애플리케이션용 그리고 문서화. 아무리 능력이 뛰어나더라도 단일 AI 에이전트가 스스로를 관리하게 될 것입니다. 과밀한 컨텍스트 창, 책임 상충 및 높은 오류 위험 계단식. 솔루션은 더 강력한 에이전트가 아닙니다. 에이전트 팀 전문가 협업하고, 서로 확인하고, 작업을 병렬화하는 것입니다.
이것이 패러다임이다 다중 에이전트 코딩: 다수의 AI 에이전트가 존재하는 시스템 자율적인 사람들은 자신의 능력을 초과하는 소프트웨어 개발 작업을 완료하기 위해 협력합니다. 단일 모델. 2025년에는 세 가지 프레임워크가 이 분야를 지배했습니다. 랭그래프 상태 저장 그래프 아키텍처를 통해 크루AI 역할 기반 모델을 통해 직관적, 전자 자동 생성 (현재 AG2) 대화식 접근 방식을 사용합니다. 이들에게 병렬 하위 에이전트가 포함된 Claude Code의 기본 시스템이 추가되었습니다.
이 문서는 아키텍처, 코드 예제 등 네 가지 시스템 모두에 대한 고급 심층 분석입니다. 귀하의 사용 사례에 적합한 것을 선택하기 위한 작업, 자세한 비교 및 실용적인 가이드입니다. 바이브 코딩의 기본 개념을 이미 잘 알고 있고 자신만의 에이전트 워크플로우를 도입하고 싶다면 다음 단계로 넘어가려면 제대로 찾아오셨습니다.
무엇을 배울 것인가
- 복잡한 개발 작업에는 단일 AI 에이전트로는 충분하지 않기 때문입니다.
- LangGraph: 그래프 아키텍처, StateGraph, 노드, 에지 및 체크포인트
- CrewAI: 역할 기반 에이전트, 작업, 순차 및 병렬 프로세스, 도구
- AutoGen/AG2: 대화형 에이전트, GroupChat, 코드 실행 샌드박스
- Claude Code: 하위 에이전트, 작업 도구 및 병렬 실행 패턴
- 세부 비교: 언제 어떤 프레임워크를 사용할지
- 실제 팀을 위한 프로덕션 준비 아키텍처
- 중요한 과제: 컨텍스트 오염, 오류 전파, 비용 관리
- 전문 에이전트 및 대체 전략에 대한 모범 사례
단일 에이전트로는 충분하지 않기 때문에
프레임워크를 살펴보기 전에, 프레임워크의 근본적인 문제를 이해하는 것이 중요합니다. 다중 에이전트 코딩이 해결됩니다. 단일 AI 에이전트에는 아무리 발전해도 한계가 있습니다. 복잡한 개발 작업에서 명확하게 나타나는 구조적 특징.
첫 번째 한계는 컨텍스트 창. 지원하는 모델이 있어도 일반적인 기업 작업에는 200,000개의 토큰이 필요하며 동시에 다음 사항을 염두에 두어야 합니다. 기존 코드베이스, 기능 사양, 아키텍처 패턴, 테스트 업데이트할 문서 및 보안 제약 조건을 작성합니다. 이 모든 것을 능가합니다 일관된 단일 컨텍스트의 기능을 신속하게 제공합니다.
두 번째 한계는 전문화. 일반 상담원은 다음과 같은 경향이 있습니다. 어떤 일을 훌륭하게 하는 것보다 모든 것을 평범한 방법으로 하는 것. 대리인 보안 전문가는 어떤 패턴을 찾아야 할지, 어떤 표준을 적용해야 할지 정확히 알고 있습니다. 테스트 전문가는 각 구성 요소 유형에 대한 테스트 패턴을 알고 있습니다.
세 번째 한계는 정확도 검토. 단일 에이전트 쓰기 코드를 작성한 다음 "테스트"하는 것은 기본적으로 동일한 코드로 작업을 확인하는 것입니다. 그가 그것을 생산한 인지적 편견. 두 개의 개별 에이전트(구현하는 에이전트와 구현하는 에이전트) reviews, bring genuinely different perspectives.
역할분리의 원칙
인간 개발팀에서는 역할(개발자, 코드 검토자, QA, 보안 엔지니어, 기술 작가)는 관료주의가 아니라 인지적 보호 장치입니다. 다중 에이전트 시스템은 AI 생성 코드에 동일한 원칙을 적용합니다. 각 에이전트의 사각지대를 체계적으로 줄여줍니다.
2025년 연구에서는 이러한 직관을 확인했습니다. AI 생성 코드의 비율은 다음과 같습니다. 감독되지 않는 단일 에이전트에 의해 생성될 경우 취약점이 상당히 높아집니다. (Veracode 2025는 인간 코드보다 2.74배 더 많은 취약점을 보고합니다.) 시스템 전담 검토 에이전트가 포함된 다중 에이전트는 이러한 격차를 크게 줄여줍니다.
LangGraph: 상태 저장 그래프 오케스트레이션
LangChain 팀이 개발한 LangGraph는 접근 방식의 자연스러운 진화를 나타냅니다. 선형 체인을 향해 순환 상태 저장 그래프. 근본적인 직관 복잡한 에이전트 워크플로는 선형이 아닙니다. 루프, 조건부 분기가 필요합니다. 한 구절과 다른 구절 사이의 상태의 평행성과 지속성.
LangChain 팀 자체는 2025년에 다음과 같이 명시적으로 전달했습니다. "에이전트에는 LangGraph를 사용하세요. 랭체인이 아닙니다." 이 선택은 에이전트 시스템이라는 중요한 아키텍처 진실을 반영합니다. 현대 시스템은 기본적으로 순차적 파이프라인이 아닌 기계였습니다.
기본 LangGraph 개념
LangGraph는 구축하기 전에 숙지해야 할 네 가지 핵심 개념을 기반으로 합니다. 효과적인 다중 에이전트 시스템:
- 상태 그래프: 워크플로 구조를 정의하는 기본 그래프 노드 간에 공유되는 상태 유형
- 상태: 정보를 나타내는 TypedDict (또는 Pydantic 모델) 그래프의 모든 노드에서 공유됨
- 노드: 상태를 수신하고 작업을 수행하는 Python 함수 (종종 LLM 호출) 및 상태 업데이트 반환
- 가장자리: 고정되거나 조건부로 설정될 수 있는 노드 간 연결 현재 상태를 기준으로
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage
import operator
# ============================================================
# 1. DEFINIZIONE DELLO STATO CONDIVISO
# ============================================================
class CodingState(TypedDict):
"""Stato condiviso tra tutti gli agenti del sistema."""
task_description: str
requirements: List[str]
generated_code: str
test_code: str
review_comments: List[str]
security_issues: List[str]
final_code: str
iteration_count: int
status: str # "planning", "coding", "testing", "reviewing", "done"
# ============================================================
# 2. INIZIALIZZAZIONE DEI MODELLI
# ============================================================
# Planner: usa Opus per ragionamento complesso
planner_model = ChatAnthropic(model="claude-opus-4-6")
# Developer: usa Sonnet per code generation veloce
developer_model = ChatAnthropic(model="claude-sonnet-4-6")
# Reviewer: usa Sonnet per analisi critica
reviewer_model = ChatAnthropic(model="claude-sonnet-4-6")
# ============================================================
# 3. DEFINIZIONE DEI NODI (AGENTI)
# ============================================================
def planner_agent(state: CodingState) -> dict:
"""Agente pianificatore: decompone il task in requirements."""
prompt = f"""Sei un software architect senior.
Task: {state['task_description']}
Analizza il task e produci una lista di requisiti tecnici specifici.
Formato: una lista puntata di requisiti chiari e implementabili.
"""
response = planner_model.invoke([HumanMessage(content=prompt)])
requirements = [
line.strip().lstrip("- ")
for line in response.content.split("\n")
if line.strip().startswith("-")
]
return {
"requirements": requirements,
"status": "coding"
}
def developer_agent(state: CodingState) -> dict:
"""Agente sviluppatore: implementa il codice dai requirements."""
requirements_text = "\n".join(f"- {r}" for r in state["requirements"])
prompt = f"""Sei un senior Python developer.
Implementa il seguente codice rispettando tutti i requisiti:
Task originale: {state['task_description']}
Requisiti:
{requirements_text}
Codice esistente (se presente):
{state.get('generated_code', 'Nessun codice precedente')}
Note del reviewer (se presenti):
{chr(10).join(state.get('review_comments', []))}
Produci SOLO il codice Python, senza spiegazioni.
"""
response = developer_model.invoke([HumanMessage(content=prompt)])
return {
"generated_code": response.content,
"status": "testing"
}
def test_writer_agent(state: CodingState) -> dict:
"""Agente test writer: scrive unit test per il codice generato."""
prompt = f"""Sei un QA engineer specializzato in Python testing.
Scrivi unit test completi per il seguente codice usando pytest:
{state['generated_code']}
Requisiti dei test:
- Copertura minima: 80%
- Testa happy path e edge cases
- Include test per casi di errore
- Usa fixtures dove appropriato
Produci SOLO il codice dei test, senza spiegazioni.
"""
response = developer_model.invoke([HumanMessage(content=prompt)])
return {
"test_code": response.content,
"status": "reviewing"
}
def code_reviewer_agent(state: CodingState) -> dict:
"""Agente code reviewer: analizza codice e test per qualità."""
prompt = f"""Sei un senior code reviewer con 10+ anni di esperienza.
Rivedi il seguente codice e i suoi test:
CODICE:
{state['generated_code']}
TEST:
{state['test_code']}
Analizza per:
1. Correttezza logica
2. qualità del codice (SOLID, DRY, KISS)
3. Completezza dei test
4. Performance
5. Leggibilita e manutenibilità
Se ci sono problemi critici, rispondimi con "NEEDS_REVISION:" seguito dai problemi.
Se il codice e accettabile, rispondimi con "APPROVED:" seguito da eventuali suggerimenti minor.
"""
response = reviewer_model.invoke([HumanMessage(content=prompt)])
comments = [response.content]
needs_revision = response.content.startswith("NEEDS_REVISION:")
return {
"review_comments": comments,
"iteration_count": state.get("iteration_count", 0) + 1,
"status": "coding" if needs_revision else "security_check"
}
def security_agent(state: CodingState) -> dict:
"""Agente sicurezza: verifica vulnerabilità nel codice."""
prompt = f"""Sei un security engineer specializzato in application security.
Analizza il seguente codice per vulnerabilità di sicurezza:
{state['generated_code']}
Cerca specificamente:
- Injection vulnerabilities (SQL, Command, Path traversal)
- Insecure deserialization
- Hardcoded secrets o credenziali
- Insecure random number generation
- Race conditions
- Input validation gaps
Lista SOLO i problemi trovati. Se nessun problema, scrivi "SECURITY_OK".
"""
response = reviewer_model.invoke([HumanMessage(content=prompt)])
issues = []
if response.content.strip() != "SECURITY_OK":
issues = [response.content]
return {
"security_issues": issues,
"final_code": state['generated_code'] if not issues else "",
"status": "done" if not issues else "coding"
}
# ============================================================
# 4. ROUTING CONDIZIONALE
# ============================================================
def route_after_review(state: CodingState) -> str:
"""Decide il prossimo nodo dopo la code review."""
if state["status"] == "coding" and state.get("iteration_count", 0) < 3:
return "developer"
elif state["status"] == "security_check":
return "security"
else:
# Max iterazioni raggiunte
return "security"
def route_after_security(state: CodingState) -> str:
"""Decide se il codice e pronto o necessità un'altra iterazione."""
if state["security_issues"] and state.get("iteration_count", 0) < 3:
return "developer"
return END
# ============================================================
# 5. COSTRUZIONE DEL GRAFO
# ============================================================
def build_coding_graph() -> StateGraph:
graph = StateGraph(CodingState)
# Aggiunta nodi
graph.add_node("planner", planner_agent)
graph.add_node("developer", developer_agent)
graph.add_node("test_writer", test_writer_agent)
graph.add_node("reviewer", code_reviewer_agent)
graph.add_node("security", security_agent)
# Entry point
graph.set_entry_point("planner")
# Edges fissi
graph.add_edge("planner", "developer")
graph.add_edge("developer", "test_writer")
graph.add_edge("test_writer", "reviewer")
# Edges condizionali
graph.add_conditional_edges(
"reviewer",
route_after_review,
{
"developer": "developer",
"security": "security"
}
)
graph.add_conditional_edges(
"security",
route_after_security,
{
"developer": "developer",
END: END
}
)
return graph
# ============================================================
# 6. ESECUZIONE CON CHECKPOINTING
# ============================================================
checkpointer = MemorySaver()
app = build_coding_graph().compile(checkpointer=checkpointer)
# Esecuzione del sistema
config = {"configurable": {"thread_id": "project-auth-001"}}
initial_state = {
"task_description": """
Implementa un sistema di autenticazione JWT in Python con:
- Login con username/password
- Generazione access token (15 min) e refresh token (7 giorni)
- Middleware per proteggere endpoint
- Revoca token (blacklist in Redis)
""",
"requirements": [],
"generated_code": "",
"test_code": "",
"review_comments": [],
"security_issues": [],
"final_code": "",
"iteration_count": 0,
"status": "planning"
}
result = app.invoke(initial_state, config=config)
print(f"Codice finale generato dopo {result['iteration_count']} iterazioni")
print(f"Problemi di sicurezza rilevati: {len(result['security_issues'])}")
이 예는 LangGraph가 복잡한 개발 워크플로를 처리하는 방법을 보여줍니다. 자동 검토 루프. 그만큼 검문소 그리고 중요한 특징 프로덕션용: 그래프 상태를 저장하고 실행을 재개할 수 있습니다. 오류가 발생한 경우 처음부터 시작하지 않고
LangGraph: 강점
- 세분화된 제어: 흐름의 모든 측면을 프로그래밍할 수 있습니다.
- 상태 지속성: 긴 워크플로우를 위한 통합 체크포인트
- 인간 참여형: 사람의 승인을 위해 일시 중지 가능성
- 병렬 실행: 노드는 병렬로 실행될 수 있습니다
- 스트리밍: 각 노드의 실시간 출력
- 테스트된 생산: 2025년 수백개 기업의 생산에 사용
CrewAI: 가상 팀을 위한 역할 기반 에이전트
CrewAI는 LangGraph에 대해 근본적으로 다른 접근 방식을 취합니다. 그래프와 상태의 관점에서 생각해 보도록 요청합니다. 팀과 역할. 각 상담원은 정의된 전문 분야와 명확한 목표를 가진 팀 구성원입니다. 그리고 사용 가능한 도구 세트. 이 정신 모델은 다음과 같은 사람들에게 더 직관적입니다. 개발팀 관리 경험이 있습니다.
2025년 버전 1.1.0에서 CrewAI는 승무원 (상위 수준의 오케스트레이션) e 흐름 (세밀한 작업 흐름 제어), 역할 기반 모델의 단순성과 함께 LangGraph의 유연성을 제공합니다.
CrewAI 아키텍처
CrewAI의 네 가지 기본 요소는 다음과 같습니다.
- 자치령 대표: 자신을 정의하는 역할, 목표 및 배경 스토리를 갖춘 AI 개체 "성격"과 전문 분야
- 작업: 설명, 예상_출력 및 특정 상담원에게 할당됨
- 도구: 추가 에이전트 기능(파일 I/O, 웹 검색, 코드 실행, 데이터베이스 액세스)
- 승무원: 프로세스를 통해 에이전트와 작업을 조정하는 팀 (순차 또는 병렬)
from crewai import Agent, Task, Crew, Process
from crewai.tools import CodeInterpreterTool, FileReadTool, FileWriteTool
from crewai_tools import GithubSearchTool
from langchain_anthropic import ChatAnthropic
# ============================================================
# 1. CONFIGURAZIONE MODELLI
# ============================================================
opus_llm = ChatAnthropic(model="claude-opus-4-6", temperature=0.1)
sonnet_llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0.2)
# ============================================================
# 2. DEFINIZIONE DEGLI STRUMENTI
# ============================================================
code_executor = CodeInterpreterTool()
file_reader = FileReadTool()
file_writer = FileWriteTool()
# ============================================================
# 3. DEFINIZIONE DEGLI AGENTI (MEMBRI DEL TEAM)
# ============================================================
tech_lead = Agent(
role="Tech Lead e Software Architect",
goal="""Analizzare i requisiti tecnici, definire l'architettura ottimale
e decomporre il lavoro in task specifici e implementabili.""",
backstory="""Hai 15 anni di esperienza in software architecture.
Hai guidato la transizione a microservizi per 3 startup unicorn.
Sei noto per la capacità di bilanciare pragmatismo e qualità tecnica.""",
llm=opus_llm,
verbose=True,
allow_delegation=True
)
senior_developer = Agent(
role="Senior Python Developer",
goal="""Implementare codice Python di alta qualità, pulito, testabile
e conforme alle best practices SOLID e alle specifiche del tech lead.""",
backstory="""Sei un Python developer con 8 anni di esperienza.
Contribuisci a progetti open source importanti.
Scrivi codice che altri developer amano leggere e mantenere.""",
llm=sonnet_llm,
verbose=True,
tools=[code_executor, file_writer],
allow_code_execution=True
)
qa_engineer = Agent(
role="QA Engineer e Test Specialist",
goal="""Verificare la correttezza del codice attraverso test exhaustivi,
trovare edge cases e garantire la copertura dei test al 90%+.""",
backstory="""Hai un background in matematica e sei ossessionato dalla
correttezza del software. Hai trovato bug critici in sistemi che erano
in produzione da anni. Non ti fidi del codice senza prove.""",
llm=sonnet_llm,
verbose=True,
tools=[code_executor],
allow_code_execution=True
)
code_reviewer = Agent(
role="Senior Code Reviewer",
goal="""Analizzare criticamente il codice per qualità, manutenibilità,
performance e conformità agli standard del team.""",
backstory="""Hai revisionato oltre 10.000 pull request nella tua carriera.
Hai un occhio infallibile per code smells, anti-pattern e problemi
di design che si manifestano solo in produzione.""",
llm=sonnet_llm,
verbose=True,
tools=[file_reader]
)
# ============================================================
# 4. DEFINIZIONE DEI TASK
# ============================================================
architecture_task = Task(
description="""
Analizza il seguente requisito e produci un documento di architettura:
REQUISITO: {task_description}
Il documento deve includere:
1. Diagramma dei componenti (in testo/ASCII)
2. Stack tecnologico raccomandato con giustificazioni
3. Struttura delle directory del progetto
4. Interfacce principali (classi/funzioni pubbliche)
5. Considerazioni su scalabilità e manutenibilità
6. Lista prioritizzata di task di implementazione
""",
expected_output="""Un documento di architettura dettagliato in markdown,
con diagrammi ASCII, struttura del progetto e lista di task.""",
agent=tech_lead
)
implementation_task = Task(
description="""
Implementa il codice Python basandoti sul documento di architettura
fornito dal Tech Lead.
Requisiti di implementazione:
- Segui esattamente le interfacce definite nell'architettura
- Usa type hints per tutte le funzioni pubbliche
- Aggiungi docstring in formato Google style
- Gestisci tutti i casi di errore con eccezioni custom
- Salva il codice in file separati per modulo
""",
expected_output="""Codice Python completo e funzionante, organizzato in moduli,
con type hints, docstring e gestione degli errori.""",
agent=senior_developer,
context=[architecture_task] # dipende dall'output dell'architettura
)
testing_task = Task(
description="""
Scrivi test completi per il codice implementato.
Requisiti:
- Framework: pytest con pytest-cov
- Copertura target: 90%
- Include unit test, integration test e test parametrici
- Testa happy path, edge cases e scenari di errore
- Usa mocking per dipendenze esterne
- Esegui i test per verificare che passino tutti
""",
expected_output="""File di test pytest completi che passano tutti,
con report di copertura al 90%+.""",
agent=qa_engineer,
context=[implementation_task]
)
review_task = Task(
description="""
Conduci una code review approfondita del codice e dei test prodotti.
Analizza per:
1. Rispetto dei principi SOLID
2. Presenza di code smells o anti-pattern
3. Correttezza logica e completezza
4. Performance e uso della memoria
5. Sicurezza (injection, validazione input, secrets)
6. qualità della documentazione
Produci un report con: problemi CRITICI, WARNINGS e SUGGESTIONS.
Per ogni problema, includi il file/riga e la fix raccomandata.
""",
expected_output="""Report di code review strutturato con categorie CRITICAL,
WARNING e SUGGESTION, con fix raccomandate per ogni problema.""",
agent=code_reviewer,
context=[implementation_task, testing_task]
)
# ============================================================
# 5. CREW - ASSEMBLAGGIO DEL TEAM
# ============================================================
coding_crew = Crew(
agents=[tech_lead, senior_developer, qa_engineer, code_reviewer],
tasks=[architecture_task, implementation_task, testing_task, review_task],
process=Process.sequential, # o Process.hierarchical per delega automatica
verbose=True,
memory=True, # abilita memoria condivisa tra agenti
embedder={
"provider": "anthropic",
"config": {"model": "claude-3-haiku-20240307"}
}
)
# ============================================================
# 6. AVVIO DEL CREW
# ============================================================
result = coding_crew.kickoff(inputs={
"task_description": """
Implementa un sistema di rate limiting per API REST con:
- Algoritmo token bucket per controllo del flusso
- Storage su Redis per distribuzione multi-instance
- Configurazione per-endpoint e per-user
- Header HTTP standard (X-RateLimit-*)
- Dashboard di monitoring con metriche Prometheus
"""
})
print(result.raw)
CrewAI 2025의 특징적인 요소는 옵션입니다. allow_code_execution=True
에이전트: 안전한 자체 관리 샌드박스에서 코드 실행이 가능합니다.
오류 및 지능적인 재시도. 코드에서 예외가 발생하면 에이전트는
오류 메시지를 확인하고 직접 해결해 보세요(최대 max_retry_limit,
기본값 2).
CrewAI: 강점
- 낮은 학습 곡선: 팀/역할 정신 모델은 직관적입니다.
- 통합 코드 실행: 자동 재시도 기능을 갖춘 보안 샌드박스
- 공유 메모리: 상담원은 이전 대화의 맥락을 기억합니다.
- MCP 지원: MCP 서버와의 양방향 통합(2025)
- 엔터프라이즈 기능: 관찰 가능성, 감사 로그, 제어 플레인 SaaS
- 흐름: 필요할 때 복잡한 작업흐름을 세밀하게 제어
AutoGen/AG2: 새로운 작업을 위한 대화형 에이전트
AutoGen(분기 이후에도 개발을 계속한 커뮤니티에 의해 AG2로 이름 변경됨) 2024년 Microsoft)는 완전히 다른 패러다임을 채택합니다. 대화의 다중 에이전트. 고정된 워크플로나 엄격한 역할을 정의하는 대신 상담원은 그들은 자연어 메시지, 새로운 행동을 통해 서로 소통합니다. 단순한 상호작용으로 인해 복잡해집니다.
이 접근 방식은 솔루션 경로가 명확하지 않은 작업에 특히 효과적입니다. 선험적으로 정의 가능: 복잡한 디버깅 문제, 연구 및 구현 혁신적인 솔루션 또는 상충되는 제약 사항 간의 협상이 필요한 작업입니다.
AG2 아키텍처: AssistantAgent 및 UserProxyAgent
AG2의 기본 빌딩 블록은 토크입니다. 보조 에이전트 (AI 에이전트) 전자 사용자프록시에이전트 (실행을 위한 프록시). UserProxyAgent는 다음을 실행할 수 있습니다. 코드를 로컬 샌드박스에 저장하여 AssistantAgent에 실제 피드백을 제공합니다. 실제 실행 결과를 기반으로 솔루션을 반복합니다.
import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from autogen.coding import LocalCommandLineCodeExecutor
import os
# ============================================================
# 1. CONFIGURAZIONE LLM
# ============================================================
config_list = [
{
"model": "claude-sonnet-4-6",
"api_key": os.environ["ANTHROPIC_API_KEY"],
"api_type": "anthropic",
}
]
llm_config = {
"config_list": config_list,
"temperature": 0.1,
"timeout": 120,
}
opus_config = {
"config_list": [{
"model": "claude-opus-4-6",
"api_key": os.environ["ANTHROPIC_API_KEY"],
"api_type": "anthropic",
}],
"temperature": 0.1,
}
# ============================================================
# 2. CODE EXECUTOR - SANDBOX LOCALE
# ============================================================
executor = LocalCommandLineCodeExecutor(
timeout=60,
work_dir="/tmp/autogen_workspace",
# Blocca comandi pericolosi
execution_policies={
"rm": False,
"curl": False,
"wget": False,
}
)
# ============================================================
# 3. DEFINIZIONE DEGLI AGENTI
# ============================================================
# Agente Software Architect - usa Opus per alta capacità di ragionamento
architect = AssistantAgent(
name="SoftwareArchitect",
system_message="""Sei un software architect senior con 15 anni di esperienza.
Il tuo ruolo e:
1. Analizzare i requisiti e proporre architetture solide
2. Definire le interfacce tra componenti
3. Identificare rischi tecnici e proporre mitigazioni
4. Guidare le decisioni tecniche del team
Quando proponi un'architettura, usa sempre diagrammi ASCII e spiega le
motivazioni delle scelte. Rispondi con "TERMINATE" quando il task e completo.""",
llm_config=opus_config,
)
# Agente Developer - implementa il codice
developer = AssistantAgent(
name="PythonDeveloper",
system_message="""Sei un senior Python developer.
Il tuo ruolo e:
1. Implementare il codice seguendo le specifiche dell'architetto
2. Scrivere codice pulito con type hints e docstring
3. Gestire gli errori in modo robusto
4. Rispettare i principi SOLID e DRY
Usa sempre blocchi di codice Python validi quando scrivi implementazioni.
Rispondi con "TERMINATE" solo quando il developer AND il reviewer approvano.""",
llm_config=llm_config,
)
# Agente Tester - scrive e verifica i test
tester = AssistantAgent(
name="QATester",
system_message="""Sei un QA engineer specializzato in Python testing.
Il tuo ruolo e:
1. Scrivere test pytest esaustivi per il codice del developer
2. Eseguire i test nella sandbox e riportare i risultati
3. Identificare edge cases non coperti
4. Garantire copertura 85%+
Produci sempre test eseguibili che posso verificare nella sandbox.""",
llm_config=llm_config,
)
# Agente Reviewer - code review critica
reviewer = AssistantAgent(
name="CodeReviewer",
system_message="""Sei un code reviewer senior con focus su qualità e sicurezza.
Il tuo ruolo e:
1. Analizzare il codice per code smells e anti-pattern
2. Verificare la sicurezza (OWASP Top 10)
3. Suggerire refactoring migliorativi
4. Approvare o richiedere modifiche
Concludi ogni review con "APPROVED" o "NEEDS_WORK: [lista problemi]".""",
llm_config=llm_config,
)
# UserProxy con esecuzione codice - fa da ponte tra agenti e sandbox
code_executor_proxy = UserProxyAgent(
name="CodeExecutor",
human_input_mode="NEVER", # completamente automatico
max_consecutive_auto_reply=5,
code_execution_config={
"executor": executor,
},
is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
)
# ============================================================
# 4. GROUP CHAT - ORCHESTRAZIONE DELLA CONVERSAZIONE
# ============================================================
group_chat = GroupChat(
agents=[architect, developer, tester, reviewer, code_executor_proxy],
messages=[],
max_round=20, # massimo 20 round di conversazione
# Strategia di selezione del prossimo parlante
speaker_selection_method="auto",
# Ordine preferito per strutturare la conversazione
allowed_or_disallowed_speaker_transitions={
architect: [developer],
developer: [tester, code_executor_proxy],
tester: [code_executor_proxy, reviewer],
code_executor_proxy: [developer, tester, reviewer],
reviewer: [developer, architect],
},
speaker_transitions_type="allowed",
)
# Manager che orchestra la group chat
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config,
system_message="""Sei il project manager di questo team di sviluppo.
Orchestri la conversazione per garantire che il task venga completato
in modo efficiente e che ogni agente contribuisca nel momento opportuno.""",
)
# ============================================================
# 5. AVVIO DELLA CONVERSAZIONE
# ============================================================
task = """
Sviluppa un sistema di caching intelligente in Python con le seguenti caratteristiche:
- Cache LRU (Least Recently Used) con TTL configurabile per entry
- Backend multipli: in-memory, Redis, o filesystem
- Decorator @cache per applicazione trasparente alle funzioni
- Statistiche di hit/miss in tempo reale
- Thread-safe per applicazioni concorrenti
Architettura, implementazione, test e code review completi.
"""
chat_result = code_executor_proxy.initiate_chat(
manager,
message=task,
summary_method="reflection_with_llm",
)
print("\n=== SOMMARIO FINALE ===")
print(chat_result.summary)
AG2: 강점
- 실제 코드 실행: 코드는 실제로 샌드박스에서 실행됩니다. 상담원은 실제 출력을 봅니다.
- 새로운 유연성: workflows adapt to the problem, 그 반대는 아니고
- AG-UI 프로토콜: 실시간 스트리밍을 갖춘 동적 프런트엔드(2025)
- 오픈텔레메트리: 에이전트 워크플로를 완벽하게 관찰할 수 있습니다.
- 인간 참여형: 쉽게 구성 가능
human_input_mode - 타입스크립트 지원: 기본 지원이 포함된 Microsoft AutoGen 버전 0.4
Claude Code: 하위 에이전트 및 병렬 실행
Claude Code는 근본적으로 다른 방식으로 작동하는 기본 다중 에이전트 시스템을 갖추고 있습니다. 지금까지 본 Python 프레임워크에서 Claude Code는 API를 통해 LLM 모델을 조정하는 대신 시작 별도의 Claude 인스턴스 (하위 에이전트) 터미널에서 직접, 각각은 자체적인 깨끗한 컨텍스트 창과 파일 시스템에서 작동하는 기능을 가지고 있습니다. 고립된 방식으로.
이 접근 방식에는 프레임워크가 없고 통합 오버헤드가 있다는 상당한 이점이 있습니다. 기본 터미널 도구 및 작업을 병렬로 실행하는 기능 다중 코어가 있는 시스템에서. 작업 도구 및 주요 메커니즘: 다음을 사용하여 하위 에이전트를 시작합니다. 특정 프롬프트를 표시하고 계속하기 전에 결과를 기다립니다(또는 작업이 독립적인 경우 모두 병렬로 시작합니다).
# Code Reviewer Agent
## Ruolo
Senior code reviewer con focus su qualità, sicurezza e manutenibilità.
## Competenze
- Principi SOLID e design patterns
- OWASP Top 10 e sicurezza applicativa
- Performance optimization
- Clean code e refactoring
## Processo di Review
1. Leggi TUTTI i file modificati prima di commentare
2. Identifica problemi per categoria: CRITICAL, WARNING, SUGGESTION
3. Per ogni problema: specifica file, riga e fix raccomandata
4. Verifica che i test coprano i casi di errore
5. Controlla presenza di hardcoded secrets o credenziali
## Output Atteso
Report strutturato in markdown con sezioni:
- Executive Summary
- CRITICAL Issues (bloccanti per il merge)
- WARNINGS (da risolvere prima della prossima release)
- SUGGESTIONS (miglioramenti opzionali)
- Security Checklist
# Security Auditor Agent
## Ruolo
Security engineer specializzato in application security e OWASP.
## Competenze
- OWASP Top 10 (2025 edition)
- Analisi di codice Python, JavaScript e SQL
- Threat modeling
- Dependency vulnerability scanning
## Checklist di Sicurezza
### Input Validation
- [ ] Tutti gli input utente sono validati e sanitizzati
- [ ] Nessuna query SQL costruita con concatenazione di stringhe
- [ ] Path traversal prevenuto con realpath/pathlib
### Authentication & Authorization
- [ ] Password hashing con bcrypt/argon2 (no MD5/SHA1)
- [ ] JWT firmati con chiavi di lunghezza adeguata
- [ ] RBAC implementato correttamente
### Secrets Management
- [ ] Nessun hardcoded secret nel codice
- [ ] Variabili d'ambiente usate per credenziali
- [ ] .gitignore include file di configurazione sensibili
### Dependencies
- [ ] Nessuna dipendenza con CVE critiche note
- [ ] requirements.txt con versioni pin-nate
## Output
- Lista vulnerabilità con CVSS score stimato
- Fix raccomandate con esempi di codice
- Priorità: CRITICAL (fix immediata), HIGH, MEDIUM, LOW
# Esempio di prompt per orchestrazione multi-agent in Claude Code
# (questo e il contenuto di un file CLAUDE.md o di un prompt interattivo)
# --- ESEMPIO 1: SEQUENTIAL MULTI-AGENT ---
# Il task tool lancia sub-agent e attende i risultati in sequenza
Implementa le seguenti feature per il modulo auth/:
1. Usa il Task tool per lanciare il sub-agent "planner":
- Input: "Analizza auth/ e crea un piano di implementazione per
aggiungere 2FA con TOTP (RFC 6238)"
- Attendi il piano completo
2. Usa il Task tool per lanciare il sub-agent "developer":
- Input: "Implementa il piano seguente: [output del planner]"
- Attendi l'implementazione completa
3. Usa il Task tool per lanciare PARALLELAMENTE:
- Sub-agent "code-reviewer": rivedi il codice in auth/
- Sub-agent "security-auditor": verifica la sicurezza di auth/
(lancia entrambi contemporaneamente, attendi entrambi)
4. Applica le fix per tutti i problemi CRITICAL trovati
---
# --- ESEMPIO 2: PARALLEL WORKTREE ISOLATION ---
# Task indipendenti su worktree separati per massima parallelizzazione
Esegui queste task IN PARALLELO su worktree separati:
Task A (worktree: feature/user-service):
- Implementa UserService con CRUD completo
- Scrivi unit test con 90% coverage
- Commit: "feat: add UserService with full CRUD"
Task B (worktree: feature/email-service):
- Implementa EmailService con template system
- Integra con SendGrid API
- Scrivi integration test
- Commit: "feat: add EmailService with SendGrid"
Task C (worktree: feature/notification-service):
- Implementa NotificationService (email + push + SMS)
- Usa UserService e EmailService come dipendenze
- Scrivi test end-to-end
- Commit: "feat: add NotificationService"
Dopo che tutti e tre i worktree sono completati:
- Mergia in ordine: A, B, C
- Risolvi eventuali conflitti
- Esegui la test suite completa
---
# --- ESEMPIO 3: AGENTE CON MEMORIA CONDIVISA ---
# Uso di file come meccanismo di comunicazione inter-agent
Prima di iniziare, crea il file /tmp/project-context.md con:
- Stack tecnologico del progetto
- Convenzioni di naming
- Pattern architetturali in uso
- Dipendenze principali
Ogni sub-agent che lanci deve:
1. Leggere /tmp/project-context.md all'inizio
2. Aggiornare /tmp/progress.md con lo stato dei propri task
3. Salvare output strutturati in /tmp/agent-outputs/[nome]/
Questo garantisce coerenza tra agenti paralleli senza conflitti di merge.
레플리트 사건의 교훈(2025)
2025년 Replit은 자율 에이전트가 제거된 사례를 문서화했습니다.
"정리" 작업 중 프로덕션 데이터베이스. 이번 사건은
명령에 대한 제한을 명시적으로 구성해야 한다는 점을 강조했습니다.
destructive. Claude Code에서 이는 섹션을 통해 처리됩니다. deny
in .claude/settings.json: 항상 차단 rm -rf,
DROP TABLE, git push --force 그리고 유사합니다.
하위 에이전트는 상위 에이전트와 동일한 제한 사항을 상속합니다.
세부 비교: 선택할 프레임워크
프레임워크의 선택은 특정 상황에 따라 다릅니다. 전체 승자는 없습니다. 각 도구는 다양한 시나리오에서 탁월합니다. 다음 표에는 차이점이 요약되어 있습니다. 결정에 도움이 되는 열쇠.
| 표준 | 랭그래프 | 크루AI | AG2(자동 생성) | 클로드 코드 |
|---|---|---|---|---|
| 어형 변화표 | 상태 저장 그래프 | 역할 기반 팀 | 이야기 잘하는 | 기본 하위 에이전트 |
| 학습 곡선 | 높음(그래프 이론) | 낮음(직관적) | 평균 | 낮은 |
| 흐름 제어 | 최고 | 중간 높음 | 신흥 | 중간 |
| 코드 실행 | 도구를 통해 | 통합(샌드박스) | 통합(로컬) | 네이티브 배쉬 |
| 상태 지속성 | 네이티브 체크포인트 | 통합 메모리 | 채팅 메시지 | 파일 시스템 |
| 병행 | 병렬 노드 | 프로세스.병렬 | 그룹챗 비동기 | 기본 작업 도구 |
| 관찰 가능성 | 랭스미스 | 크루AI 엔터프라이즈 | 오픈텔레메트리 | 네이티브 로그 |
| 인간 참여형 | 네이티브 인터럽트 | 콜백 | human_input_mode |
인터랙티브 |
| 생태계 | 랭체인 | 독립적인 | 마이크로소프트/AG2 | 인류학 |
| 다음에 이상적입니다. | 조건부 논리를 사용하는 복잡한 워크플로 | 역할이 정의된 가상 팀 | 코드 실행을 통한 탐색 작업 | CLI 자동화 및 기존 프로젝트 |
실용적인 권장 사항
- 랭그래프를 선택하세요 포크가 많은 복잡한 워크플로가 있는 경우 조건부, 세션 간 안정적인 상태 지속성이 필요하고 팀이 있습니다. 상태 머신과 유향 그래프의 개념을 이해하는 사람.
- CrewAI를 선택하세요 멀티 에이전트를 빠르게 시작하고 싶다면, 귀하의 팀은 조직의 역할 측면에서 생각하는 데 익숙하며 다음이 필요합니다. 자동 오류 관리 기능을 갖춘 통합 코드 실행.
- AG2를 선택하세요 탐색적 작업이고 사전 정의된 워크플로가 없는 경우 실제 코드 실행과 함께 실제 다각적인 대화가 필요합니다. 귀하는 Microsoft 생태계(Azure, TypeScript)에 있습니다.
- 클로드 코드를 선택하세요 이미 Claude와 협력하고 있다면 다중 에이전트를 원할 것입니다. 추가 프레임워크 없이 주요 사용 사례 및 작업 자동화 기존 코드베이스에서 개발.
실제 팀을 위한 프로덕션 지원 아키텍처
프로덕션 중인 다중 에이전트 시스템에는 다양한 아키텍처 고려 사항이 필요합니다. 단순한 에이전트 오케스트레이션을 넘어 이를 배포한 회사는 2025년 시스템은 미리 알아두면 유용한 몇 가지 중요한 교훈을 배웠습니다. 시작하려면.
통합 아키텍처 패턴
2025년 연구에 따르면 다중 에이전트 코딩에 가장 효과적인 패턴은 생산은 세 가지입니다:
- 감독자 패턴: 오케스트레이터 에이전트는 에이전트에 작업을 위임합니다. 결과를 전문화하고 집계합니다. 조건부 라우팅을 사용하는 LangGraph에 이상적입니다.
- 파이프라인 패턴: 하나의 출력이 순서대로 에이전트
다음의 입력이 됩니다. CrewAI의 자연스러움
context및 작업 직원. - 피어 투 피어 패턴: 서로 자유롭게 소통하는 에이전트 중앙 오케스트레이터 없이. GroupChat을 사용하면 AG2에서 자연스러워집니다.
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from typing import TypedDict, Optional
import logging
import time
logger = logging.getLogger(__name__)
class ProductionCodingState(TypedDict):
task_id: str
task_description: str
generated_code: str
test_results: str
review_status: str
error_count: int
max_errors: int
last_error: Optional[str]
start_time: float
max_execution_time: float # secondi
# ============================================================
# ERROR HANDLING - Nodo dedicato alla gestione errori
# ============================================================
def error_handler(state: ProductionCodingState) -> dict:
"""Gestisce errori e decide se riprovare o terminare."""
error_count = state.get("error_count", 0) + 1
logger.error(
f"Task {state['task_id']}: errore #{error_count} - {state.get('last_error', 'unknown')}"
)
if error_count >= state["max_errors"]:
logger.critical(f"Task {state['task_id']}: max errori raggiunto, terminazione")
return {"error_count": error_count, "review_status": "FAILED"}
# Exponential backoff prima del retry
wait_time = min(2 ** error_count, 30)
logger.info(f"Retry in {wait_time}s...")
time.sleep(wait_time)
return {"error_count": error_count, "review_status": "RETRY"}
# ============================================================
# TIMEOUT MANAGEMENT
# ============================================================
def check_timeout(state: ProductionCodingState) -> str:
"""Verifica se il task ha superato il timeout massimo."""
elapsed = time.time() - state["start_time"]
if elapsed > state["max_execution_time"]:
logger.warning(
f"Task {state['task_id']}: timeout dopo {elapsed:.1f}s "
f"(max: {state['max_execution_time']}s)"
)
return "timeout"
return "continue"
# ============================================================
# COST TRACKING - Tracciamento costi in tempo reale
# ============================================================
class CostTracker:
# Prezzi Anthropic (USD per 1M token, Feb 2026)
PRICES = {
"claude-opus-4-6": {"input": 15.0, "output": 75.0},
"claude-sonnet-4-6": {"input": 3.0, "output": 15.0},
"claude-haiku-4-5": {"input": 0.25, "output": 1.25},
}
def __init__(self, budget_usd: float):
self.budget = budget_usd
self.spent = 0.0
self.calls = 0
def track(self, model: str, input_tokens: int, output_tokens: int) -> None:
prices = self.PRICES.get(model, {"input": 3.0, "output": 15.0})
cost = (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000
self.spent += cost
self.calls += 1
if self.spent > self.budget * 0.8:
logger.warning(f"Costo task: ${self.spent:.4f} (80% del budget ${self.budget})")
def check_budget(self) -> bool:
"""Ritorna True se il budget e ancora disponibile."""
return self.spent < self.budget
def report(self) -> dict:
return {
"total_cost_usd": round(self.spent, 4),
"budget_remaining_usd": round(self.budget - self.spent, 4),
"api_calls": self.calls,
"budget_utilization_pct": round(self.spent / self.budget * 100, 1)
}
# ============================================================
# PERSISTENZA PRODUCTION CON POSTGRESQL
# ============================================================
# Il checkpointing su PostgreSQL garantisce durabilita del workflow
# anche in caso di crash dell'applicazione
def create_production_app(db_connection_string: str):
"""Crea un'applicazione LangGraph con persistenza PostgreSQL."""
# PostgresSaver per persistenza production-grade
checkpointer = PostgresSaver.from_conn_string(db_connection_string)
checkpointer.setup() # crea le tabelle se non esistono
graph = StateGraph(ProductionCodingState)
# ... aggiunta nodi e edges ...
return graph.compile(checkpointer=checkpointer)
# ============================================================
# CIRCUIT BREAKER - Protezione contro failure a cascata
# ============================================================
class CircuitBreaker:
"""Implementa il Circuit Breaker pattern per agenti inaffidabili."""
def __init__(self, failure_threshold: int = 5, recovery_time: int = 60):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.recovery_time = recovery_time
self.last_failure_time = 0
self.state = "CLOSED" # CLOSED=normale, OPEN=blocco, HALF-OPEN=test
def call(self, func, *args, **kwargs):
"""Chiama la funzione attraverso il circuit breaker."""
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.recovery_time:
self.state = "HALF-OPEN"
logger.info("Circuit breaker: passaggio a HALF-OPEN")
else:
raise Exception("Circuit breaker OPEN: servizio non disponibile")
try:
result = func(*args, **kwargs)
if self.state == "HALF-OPEN":
self.state = "CLOSED"
self.failure_count = 0
logger.info("Circuit breaker: recupero riuscito, ritorno a CLOSED")
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
logger.error(f"Circuit breaker OPEN dopo {self.failure_count} errori")
raise
다중 에이전트 코딩의 과제와 한계
프로덕션의 다중 에이전트 시스템은 단지 등장하기만 하는 명백하지 않은 문제를 제시합니다. 경험이 있습니다. 이를 미리 아는 것이 견고한 시스템의 차이입니다. 그리고 최악의 시간에 생산이 중단되는 것입니다.
5가지 주요 과제
- 상황 오염: 에이전트가 출력에 정보를 포함하는 경우 후속 에이전트에 부정적인 영향을 미치는 관련성이 없거나 시끄러운 내용입니다. 해결책: 자유 텍스트 대신 구조화된 스키마 출력(JSON/Pydantic)을 정의합니다.
- 오류 전파: 업스트림 에이전트의 오류가 증폭되었습니다. 파이프라인을 통해 점점 진실과 멀어지는 출력을 생성합니다. 해결책: 스키마 유효성 검사 및 대체를 통해 각 단계에서 출력 유효성을 검사합니다.
- 비용 폭발: 수렴하지 않는 검토 루프가 생성될 수 있습니다. 수백 건의 API 호출과 작업당 수십 달러의 비용이 발생합니다. 해결책: 하드 한도 및 회로 차단기를 사용한 예산 추적.
- 에이전트 간 불일치: 두 명의 상담원이 서로 다른 고용을 할 수 있음 동일한 아키텍처 지점에서 일관되지 않은 코드를 생성합니다. 해결책: 공유 명시적 상태 또는 공유 컨텍스트 문서.
- 복잡성 모니터링: 에이전트가 5개를 초과하면 디버깅이 기하급수적으로 더 복잡해졌습니다. 솔루션: 상관 ID를 사용한 분산 추적 각 작업별로 표준화된 로그 구조를 제공합니다.
명심해야 할 중요한 통계: 2025년 연구에 따르면 다중 에이전트 시스템의 75%가 초과되면 관리하기 어려워집니다. 주로 디버깅 복잡성의 기하급수적인 증가로 인해 에이전트 5개 그리고 모니터링. 항상 필요한 최소 상담원 수로 시작하세요.
효과적인 다중 에이전트 시스템을 위한 모범 사례
1. 에이전트의 수술 전문화
모든 에이전트는 다음을 갖추어야 합니다. 하나의 주요 책임 그리고 세트
그것을 수행하는 데 필요한 최소한의 도구. 모든 일을 실제로 수행하는 에이전트
잘하는 게 하나도 없는 에이전트. 정의에서는 system_message 또는
의 backstory, 구체적으로 설명하세요. "당신은 다음 분야를 전문으로 하는 보안 엔지니어입니다.
OWASP Top 10'이며 '당신은 보안 전문가입니까?'보다 훨씬 더 효과적입니다.
2. 구조화된 스키마 출력
Pydantic 모델 또는 JSON 스키마를 사용하여 각 에이전트에서 예상되는 출력을 정의합니다.
자유 텍스트는 파이프라인을 통해 증폭되는 모호성을 도입합니다.
생산하는 에이전트 ReviewReport 필드가 있는
critical_issues: List[Issue] 그리고 하나보다 무한히 더 신뢰할 수 있습니다
다음 에이전트가 해석할 수 있도록 구조화되지 않은 텍스트를 생성합니다.
3. 모니터링 및 관찰 가능성
프로덕션에서는 각 에이전트 호출을 타임스탬프, 사용된 모델, 입력/출력 토큰, 비용, 대기 시간 및 상태. LangGraph 사용의 경우 랭스미스; CrewAI의 경우 기업 관찰 가능성을 사용합니다. AG2의 경우 OpenTelemetry를 사용합니다. 사용자 정의 시스템의 경우 항상 상관 관계 ID를 사용하여 로그를 구성하십시오.
4. 대체 및 우아한 성능 저하
각 중요 에이전트에는 폴백이 있어야 합니다. 즉, 더 간단한 에이전트(+ 모델) 경제적), 캐시된 응답 또는 인적 에스컬레이션 경로 등이 있습니다. 시스템은 다음과 같아야 합니다. 에이전트가 실패하더라도 성능이 저하된 상태에서도 계속 작동합니다.
5. 강제 수렴을 통한 제한된 반복
모든 검토 루프에는 최대 반복 제한이 있어야 합니다.
이론적으로 수렴을 보장할 수 없습니다. 리뷰어가 묻는 질문
늘 변화하고, 그것을 항상 가볍게 구현하는 개발자
그렇지 않으면 무한정 실행될 수 있습니다. 항상 설정 max_iterations
한계에 도달하면 "가용 가능한 최상의" 것을 수락합니다.
from pydantic import BaseModel, Field
from typing import List, Literal, Optional
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
# ============================================================
# OUTPUT SCHEMA STRUTTURATI - Eliminano ambiguità tra agenti
# ============================================================
class Issue(BaseModel):
"""Singolo problema identificato nella code review."""
file_path: str = Field(description="Path del file con il problema")
line_number: Optional[int] = Field(default=None, description="Numero di riga")
severity: Literal["CRITICAL", "HIGH", "MEDIUM", "LOW"]
category: Literal["security", "logic", "performance", "style", "test_coverage"]
description: str = Field(description="Descrizione chiara del problema")
suggested_fix: str = Field(description="Fix raccomandata con esempio di codice")
class CodeReviewReport(BaseModel):
"""Report strutturato di code review."""
approved: bool = Field(description="True se il codice può essere mergato")
summary: str = Field(description="Sommario esecutivo in 2-3 frasi")
critical_issues: List[Issue] = Field(default_factory=list)
warnings: List[Issue] = Field(default_factory=list)
suggestions: List[Issue] = Field(default_factory=list)
security_score: int = Field(ge=0, le=100, description="Score sicurezza 0-100")
quality_score: int = Field(ge=0, le=100, description="Score qualità 0-100")
test_coverage_estimate: float = Field(
ge=0.0, le=1.0,
description="Stima copertura test (0.0-1.0)"
)
class SecurityAuditReport(BaseModel):
"""Report di security audit strutturato."""
has_critical_vulnerabilities: bool
owasp_findings: List[Issue] = Field(default_factory=list)
hardcoded_secrets: List[str] = Field(default_factory=list)
vulnerable_dependencies: List[str] = Field(default_factory=list)
remediation_priority: Literal["immediate", "high", "medium", "low"]
cvss_score_estimate: float = Field(ge=0.0, le=10.0)
# ============================================================
# UTILIZZO: agente con output strutturato
# ============================================================
def code_reviewer_with_schema(code: str) -> CodeReviewReport:
"""Agente code reviewer che produce output strutturato."""
model = ChatAnthropic(model="claude-sonnet-4-6")
# Usa structured output per garantire formato corretto
structured_model = model.with_structured_output(CodeReviewReport)
prompt = f"""Conduci una code review approfondita del seguente codice Python:
{code}
Analizza per: correttezza logica, qualità, sicurezza, performance e copertura test."""
report = structured_model.invoke([HumanMessage(content=prompt)])
return report
# ============================================================
# UTILIZZO NEL GRAFO LANGGRAPH
# ============================================================
def reviewer_node(state: dict) -> dict:
"""Nodo LangGraph che usa output strutturato."""
report = code_reviewer_with_schema(state["generated_code"])
return {
"review_approved": report.approved,
"critical_issues": [issue.model_dump() for issue in report.critical_issues],
"security_score": report.security_score,
"quality_score": report.quality_score,
# Il prossimo agente riceve dati strutturati, non testo ambiguo
"review_report": report.model_dump()
}
결론: 2026년 다중 에이전트 시스템의 미래
2025년 다중 에이전트 코딩은 개념 증명에서 도구로 성숙해졌습니다. 생산 준비. LangGraph는 복잡하고 통제된 작업흐름을 지배합니다. 역할이 정의된 가상 팀을 위한 CrewAI, 탐색 작업을 위한 AG2 실제 코드 실행 및 생태계의 기본 자동화를 위한 Claude Code 인류.
2026년 가장 중요한 트렌드는 프로토콜 표준화 에이전트 간: MCP(Model Context Protocol)와 A2A(에이전트 대 에이전트)가 등장하고 있습니다. 서로 다른 프레임워크의 에이전트가 서로 통신할 수 있게 해주는 사실상의 표준입니다. LangGraph 에이전트가 CrewAI 전문가를 부를 수 있는 멀지 않은 미래 AG2 도구를 사용하는 것은 공상 과학 소설이 아닙니다. 이미 부분적으로 가능합니다.
즉시 시작하고 싶은 개발자를 위한 조언은 실용적입니다. 시작하지 마세요. 가장 정교한 프레임워크에서 두 명의 에이전트(개발자 + 검토자)로 시작하세요. 자신이 가장 잘 아는 도구를 사용하여 결과를 측정하고 복잡성만 추가하면 됩니다. 문제가 정말로 필요할 때. 소프트웨어 엔지니어링의 황금률 여기에도 적용됩니다. 문제를 해결하는 가장 좋고 간단한 다중 에이전트 시스템 당신의 문제.
권장되는 다음 단계
- 이전 기사: 법률 에이전트 워크플로: AI 문제 분해 작업을 병렬화하기 전에 작업을 구조화하는 방법을 이해합니다.
- 다음 기사: 가다 AI 생성 코드 테스트 AI 생성 코드와 관련된 품질 보증 전략
- 안전: 법률 Vibe 코딩의 보안 다중 에이전트 시스템의 취약점을 관리하기 위해
- 클로드 특정: 자세히 알아보기 클로드 코드: 터미널 에이전트 개발 기본 하위 에이전트 시스템에 대한 자세한 내용은
Vibe Coding 및 Agentic 개발 시리즈
이 시리즈의 네 번째 기사입니다. 이전 기사 내용 바이브코딩 패러다임, 클로드코드를 터미널에서 분해, 에이전트 워크플로. 향후 기사에서는 AI 생성 코드 테스트, IDE를 위한 신속한 엔지니어링, 바이브 코딩 보안, 개발의 미래 2026년에 에이전트.







