그래프 엔지니어링: 하나의 프롬프트로 1,000 개의 AI 에이전트를 병렬 실행하는 방법

1시도해 보기 → 결과 확인 → 조정 → 다시 시도

1"이 코드베이스를 읽고 변경 로그를 작성하세요"
2"가격 페이지를 가져와서 경쟁사 기능을 요약하세요"

1from dataclasses import dataclass
2
3@dataclass
4class TaskNode:
5    id: str
6    prompt: str
7    depends_on: list[str]  # 이 노드가 실제로 필요로 하는 노드의 ID
8
9def has_real_edge(node_a: TaskNode, node_b: TaskNode) -> bool:
10    """
11    핵심 그래프 엔지니어링 테스트:
12    node_b의 프롬프트가 실제로 node_a의 출력을 필요로 하는가?
13    """
14    return node_a.id in node_b.depends_on
15
16# 예시: 대부분의 "체인"은 2-3개의 실제 의존성 그룹으로 축소됩니다
17nodes = [
18    TaskNode("audit_routes", "모든 API 라우트 파일 나열", []),
19    TaskNode("check_auth", "인증 미들웨어 적용 범위 확인", []),
20    TaskNode("fetch_weather", "오늘 날씨 가져오기", []),
21    TaskNode("summarize", "라우트 + 인증 결과 요약",
22              depends_on=["audit_routes", "check_auth"]),
23]
24
25# audit_routes, check_auth, fetch_weather 사이에는 엣지가 없습니다
26# 병렬로 실행됩니다. "summarize"만 실제 엣지를 가지고 있습니다 -- 기다립니다.

1이 코드베이스의 모든 라우트 파일을 감사하는 워크플로우를 생성하세요.
2
3각 라우트 파일에 대해 독립적으로 확인하세요:
4- 인증 미들웨어 존재 여부
5- 모든 매개변수에 대한 입력 검증
6- 속도 제한 설정 여부
7- 오류 처리에서 스택 추적이 유출되지 않는지
8
9모든 라우트 파일에 대해 이러한 검사를 병렬로 실행하세요 —
10서로 의존하지 않습니다.
11
12모든 파일이 확인된 후, 심각도별로 그룹화된 하나의 통합
13보고서를 생성하세요: 치명적, 경고, 정보.
14
15통합 단계는 모든 검사가 완료될 때까지 기다려야 합니다.
16그 이전의 모든 단계는 기다리면 안 됩니다.

1import asyncio
2from anthropic import Anthropic
3
4client = Anthropic()
5
6async def audit_route_file(filepath: str) -> dict:
7    """하나의 노드. 다른 모든 라우트 파일과 독립적으로 실행됩니다."""
8    response = await client.messages.create(
9        model="claude-sonnet-5",
10        max_tokens=1000,
11        messages=[{
12            "role": "user",
13            "content": f"""이 라우트 파일을 감사하세요:
14            - 인증 미들웨어, 입력 검증,
15              속도 제한, 오류 처리
16
17            파일: {filepath}
18
19            JSON 반환: {{"file": "", "issues": [], "severity": ""}}"""
20        }]
21    )
22    return {"file": filepath, "result": response.content[0].text}
23
24async def consolidate(results: list[dict]) -> str:
25    """하나의 실제 엣지 -- 모든 감사 노드가 완료될 때까지 기다립니다."""
26    response = await client.messages.create(
27        model="claude-opus-4-8",
28        max_tokens=2000,
29        messages=[{
30            "role": "user",
31            "content": f"""다음 {len(results)}개의 라우트 감사 결과를
32            심각도별로 그룹화된 하나의 보고서로 통합하세요:
33
34            {results}"""
35        }]
36    )
37    return response.content[0].text
38
39async def run_graph(route_files: list[str]):
40    # 확산 -- 모든 독립 노드가 동시에 실행됩니다
41    audit_tasks = [audit_route_file(f) for f in route_files]
42    results = await asyncio.gather(*audit_tasks)
43
44    # 수렴 -- 실제 의존성이 있는 하나의 노드
45    report = await consolidate(results)
46    return report
47
48# 40개의 라우트 파일, 하나의 프롬프트, 하나의 병렬 패스
49results = asyncio.run(run_graph([
50    f"routes/{f}.py" for f in ["auth", "users", "billing", "orders"]
51    # ...36개 더

1async def layered_consolidate(results: list[dict], batch_size: int = 30):
2    """계층별 수렴 -- 대규모 원시 출력을 절대 합성하지 않습니다."""
3    batches = [results[i:i+batch_size]
4               for i in range(0, len(results), batch_size)]
5
6    batch_summaries = await asyncio.gather(*[
7        summarize_batch(batch) for batch in batches
8    ])
9
10    # 최종 통합은 1,000개의 원시 결과가 아닌 요약을 대상으로 작업합니다
11    return await consolidate(batch_summaries)

1async def safe_consolidate(results: list[dict], expected_count: int):
2    if len(results) < expected_count:
3        missing = expected_count - len(results)
4        print(f"경고: {missing}개 노드가 무음 실패했습니다. "
5              f"보고서가 불완전할 수 있습니다.")
6    return await consolidate(results)

1                    오케스트레이터
2                         |
3        +--------+-------+-------+--------+
4        v        v       v       v        v
5     노드 1   노드 2  노드 3  ...      노드 N
6   (병렬, 노드 간 엣지 없음)
7        |        |       |               |
8        +--------+-------+-------+-------+
9                         v
10                 배치 요약  <- 계층별 수렴
11                 (30개씩 그룹화)
12                         v
13                 최종 보고서   <- 하나의 진정한 엣지

1async def orchestrate(task: str, resources: list[str]):
2    """
3    오케스트레이터 노드 -- 분해만 하고 실행하지 않습니다.
4    """
5    plan = await client.messages.create(
6        model="claude-opus-4-8",
7        max_tokens=2000,
8        messages=[{
9            "role": "user",
10            "content": f"""작업: {task}
11            사용 가능한 리소스: {resources}
12
13            그래프로 분해:
14            - 각 독립 노드 나열 (공유 엣지 없음)
15            - 노드 간 실제 의존성 나열
16            - 개수가 50개를 초과하는 경우 노드를 수렴 배치로 그룹화
17
18            다음을 포함하는 JSON 반환: nodes, edges, batch_groups"""
19        }]
20    )
21
22    graph = parse_plan(plan.content[0].text)
23
24    # 독립 노드를 병렬로 실행
25    node_results = await asyncio.gather(*[
26        execute_node(n) for n in graph["nodes"] if not n["depends_on"]
27    ])
28
29    # 그런 다음 실제 엣지만 존중하여 의존 노드 실행
30    final = await execute_dependent_chain(graph["edges"], node_results)
31
32    return final