그래프 09. 병렬 리서치 + Judge 품질 루프 — 종합 예제

# 07, 08 튜토리얼 완료 확인
euleragent graph validate examples/graphs/parallel/my_first_parallel.yaml
euleragent graph validate examples/graphs/parallel/three_branch_research.yaml

mkdir -p examples/graphs/capstone

plan
  │
  ├──→ web_search   (plan mode + HITL) ──┐
  ├──→ local_search (execute mode)    ──┤──→ merge_findings → evaluate ⇄ revise
  └──→ doc_search   (execute mode)    ──┘                          ↓ finalize

state_schema:
  # 3개 브랜치가 모두 수집 결과를 합산
  findings:
    type: list
    merge: append_list   # web_search, local_search, doc_search 모두 씀 → 결정론적

  # 각 브랜치가 찾은 소스 수를 합산
  source_count:
    type: integer
    merge: sum_int       # 3개 브랜치 모두 씀 → 결정론적

  # merge_findings가 작성하는 최종 요약 (단일 노드)
  final_summary:
    type: string
    merge: last_write    # merge_findings만 씀 → 안전

parallel_groups:
  - id: research_group
    branches: [web_search, local_search, doc_search]
    join: merge_findings

- id: plan
  kind: llm
  runner:
    mode: execute      # 검색 방향 결정 (자율 실행)
  artifacts:
    primary: search_plan.md

- id: web_search
  kind: llm
  runner:
    mode: plan             # HITL: 검색 계획 제안 → 사람 승인 → 실행
    force_tool: web.search # 반드시 web.search 도구 사용
    min_proposals: 2       # 최소 2개 검색 쿼리 제안
  guardrails:
    tool_call_budget:
      web.search: 5        # 웹 검색 최대 5회
  writes_state: [findings, source_count]

- id: local_search
  kind: llm
  runner:
    mode: execute          # 자율 실행 (승인 불필요)
    exclude_tools: [shell.exec, file.write]  # 안전 설정
  writes_state: [findings, source_count]

- id: doc_search
  kind: llm
  runner:
    mode: execute
    exclude_tools: [shell.exec, file.write, web.search]
  writes_state: [findings]   # source_count는 쓰지 않음 (문서 수는 별도 집계)

- id: merge_findings
  kind: llm
  runner:
    mode: execute
  reads_state: [findings, source_count]
  writes_state: [final_summary]
  artifacts:
    primary: research_report.md

- id: evaluate
  kind: judge
  judge:
    schema: evaluator_v1
    route_values: [finalize, revise]

- id: revise
  kind: llm
  runner:
    mode: execute
    max_loops: 2
  artifacts:
    primary: research_report.md

edges:
  # plan → 팬아웃 (3개 브랜치 동시 시작)
  - from: plan
    to: web_search
    when: "true"
  - from: plan
    to: local_search
    when: "true"
  - from: plan
    to: doc_search
    when: "true"

  # 팬인: web_search → merge_findings
  # web_search는 HITL이므로 approvals_resolved 사용
  - from: web_search
    to: merge_findings
    when: "approvals_resolved"   # ← HITL 승인 완료 후 조인

  # 팬인: local_search, doc_search → merge_findings
  # 자율 실행 브랜치는 "true" 사용
  - from: local_search
    to: merge_findings
    when: "true"
  - from: doc_search
    to: merge_findings
    when: "true"

  # 하류 순차 실행
  - from: merge_findings
    to: evaluate
    when: "true"
  - from: evaluate
    to: finalize
    when: "judge.route == finalize"
  - from: evaluate
    to: revise
    when: "judge.route == revise"
  - from: revise
    to: evaluate
    when: "true"

web_search (HITL) ──[approvals_resolved]──┐
local_search      ──[true]────────────────┤──→ merge_findings
doc_search        ──[true]────────────────┘

LangGraph는 3개의 팬인 엣지를 모두 기다린 후 merge_findings를 실행합니다.
web_search가 HITL로 지연되어도 LangGraph가 모든 브랜치 완료를 기다립니다.

# examples/graphs/capstone/parallel_research_with_quality.yaml
id: graph.parallel_research_with_quality
version: 1
category: research
description: |
  3개 소스 병렬 리서치 + Judge 품질 루프
  plan → [web_search(HITL) | local_search | doc_search] → merge_findings → evaluate ⇄ revise

state_schema:
  findings:
    type: list
    merge: append_list

  source_count:
    type: integer
    merge: sum_int

  final_summary:
    type: string
    merge: last_write

defaults:
  max_iterations: 4
  max_total_tool_calls: 60
  max_web_search_calls: 10

parallel_groups:
  - id: research_group
    branches: [web_search, local_search, doc_search]
    join: merge_findings

nodes:
  - id: plan
    kind: llm
    runner:
      mode: execute
    artifacts:
      primary: search_plan.md

  - id: web_search
    kind: llm
    runner:
      mode: plan
      force_tool: web.search
      min_proposals: 2
    guardrails:
      tool_call_budget:
        web.search: 5
    writes_state: [findings, source_count]

  - id: local_search
    kind: llm
    runner:
      mode: execute
      exclude_tools: [shell.exec, file.write]
    writes_state: [findings, source_count]

  - id: doc_search
    kind: llm
    runner:
      mode: execute
      exclude_tools: [shell.exec, file.write, web.search]
    writes_state: [findings]

  - id: merge_findings
    kind: llm
    runner:
      mode: execute
    reads_state: [findings, source_count]
    writes_state: [final_summary]
    artifacts:
      primary: research_report.md

  - id: evaluate
    kind: judge
    judge:
      schema: evaluator_v1
      route_values: [finalize, revise]

  - id: revise
    kind: llm
    runner:
      mode: execute
      max_loops: 2
    artifacts:
      primary: research_report.md

edges:
  # plan → 팬아웃
  - from: plan
    to: web_search
    when: "true"
  - from: plan
    to: local_search
    when: "true"
  - from: plan
    to: doc_search
    when: "true"

  # 팬인
  - from: web_search
    to: merge_findings
    when: "approvals_resolved"   # HITL 승인 후
  - from: local_search
    to: merge_findings
    when: "true"
  - from: doc_search
    to: merge_findings
    when: "true"

  # 하류 순차
  - from: merge_findings
    to: evaluate
    when: "true"
  - from: evaluate
    to: finalize
    when: "judge.route == finalize"
  - from: evaluate
    to: revise
    when: "judge.route == revise"
  - from: revise
    to: evaluate
    when: "true"

finalize:
  artifact: research_report.md

euleragent graph validate examples/graphs/capstone/parallel_research_with_quality.yaml

검증 중: examples/graphs/capstone/parallel_research_with_quality.yaml

단계 1/3: YAML 파싱...
  id: graph.parallel_research_with_quality
  노드 수: 7 (plan, web_search, local_search, doc_search, merge_findings, evaluate, revise)
  엣지 수: 10
  parallel_groups: 1개 (research_group, 3개 브랜치)
  완료

단계 2/3: Pattern 기본 검증...
  [✓] 노드 ID 유일성
  [✓] 엣지 소스/타겟 존재
  [✓] finalize 도달 가능성
  [✓] judge route_values 커버리지
        evaluate: [finalize, revise] ← 엣지와 일치 ✓
  [✓] 순환 감지됨: evaluate → revise → evaluate
      max_iterations: 4 ✓
  완료

단계 3/3: Graph 추가 검증...
  [✓] state_schema 존재
  [✓] state_schema 타입+merge 호환성:
        findings: list + append_list ✓
        source_count: integer + sum_int ✓
        final_summary: string + last_write ✓
  [✓] parallel_groups 검증:
        research_group:
          branches: [web_search, local_search, doc_search] (3개)
          join: merge_findings ✓
  [✓] 각 브랜치 writes_state:
        web_search: [findings, source_count] ✓
        local_search: [findings, source_count] ✓
        doc_search: [findings] ✓  (source_count 미선언 — 허용)
  [✓] findings (append_list): 3개 브랜치 사용 → 결정론적 ✓
  [✓] source_count (sum_int): 2개 브랜치 사용 → 결정론적 ✓
  [✓] 팬아웃 엣지: plan → web_search, local_search, doc_search ✓
  [✓] 팬인 엣지:
        web_search → merge_findings (approvals_resolved) ✓
        local_search → merge_findings (true) ✓
        doc_search → merge_findings (true) ✓
  [✓] 브랜치가 finalize로 직접 라우팅 안 함 ✓
  [✓] 병렬 브랜치 부작용 도구:
        web_search: force_tool=web.search (OK, web.search는 허용)
        local_search: exclude_tools=[shell.exec, file.write] ✓
        doc_search: exclude_tools=[shell.exec, file.write, web.search] ✓
  완료

결과: 유효 ✓ (오류 없음, 경고 없음)

euleragent graph compile \
  examples/graphs/capstone/parallel_research_with_quality.yaml \
  --out examples/graphs/capstone/parallel_research_compiled.json

# IR 전체 구조 요약 출력
python -m json.tool examples/graphs/capstone/parallel_research_compiled.json | \
  python -c "
import sys, json
d = json.load(sys.stdin)
print(f'graph_type: {d[\"graph_type\"]}')
print(f'id: {d[\"id\"]}')
print(f'compiled_at: {d[\"compiled_at\"]}')
print()
print('=== state_schema ===')
for key, val in d['state_schema'].items():
    print(f'  {key}: {val[\"type\"]} + {val[\"merge\"]}')
print()
print('=== parallel_groups ===')
for g in d['parallel_groups']:
    print(f'  {g[\"id\"]}: branches={g[\"branches\"]}, join={g[\"join\"]}')
print()
print('=== nodes ===')
for n in d['nodes']:
    ib = n.get('interrupt_before', False)
    ia = n.get('interrupt_after', False)
    ws = n.get('writes_state', [])
    rs = n.get('reads_state', [])
    print(f'  [{n[\"kind\"]}] {n[\"id\"]} writes={ws} reads={rs} ib={ib} ia={ia}')
print()
print('=== langgraph_builder ===')
lb = d['langgraph_builder']
print(f'  interrupt_before: {lb[\"interrupt_before\"]}')
print(f'  interrupt_after: {lb[\"interrupt_after\"]}')
print(f'  conditional_edges: {len(lb[\"add_conditional_edges\"])}개')
"

graph_type: graph
id: graph.parallel_research_with_quality
compiled_at: 2026-02-23T11:00:00Z

=== state_schema ===
  findings: list + append_list
  source_count: integer + sum_int
  final_summary: string + last_write

=== parallel_groups ===
  research_group: branches=['web_search', 'local_search', 'doc_search'], join=merge_findings

=== nodes ===
  [llm] plan writes=[] reads=[] ib=False ia=False
  [llm] web_search writes=['findings', 'source_count'] reads=[] ib=False ia=False
  [llm] local_search writes=['findings', 'source_count'] reads=[] ib=False ia=False
  [llm] doc_search writes=['findings'] reads=[] ib=False ia=False
  [llm] merge_findings writes=['final_summary'] reads=['findings', 'source_count'] ib=False ia=False
  [judge] evaluate writes=[] reads=[] ib=False ia=False
  [llm] revise writes=[] reads=[] ib=False ia=False

=== langgraph_builder ===
  interrupt_before: []
  interrupt_after: []
  conditional_edges: 1개

strategy_plan
    │
    ├──→ [competitor_a | competitor_b | competitor_c] → merge_competitors
    │    (parallel_group_1: competitor_research)
    │
    └──→ market_analysis (순차, 병렬 아님)
         ↓
    gap_analysis (merge_competitors + market_analysis 결과 모두 읽기)
         ↓
    positioning_draft
         ↓
    quality_judge ⇄ positioning_revise
         ↓ finalize

# 작성 후 전체 검증
euleragent graph validate examples/graphs/capstone/competitor_positioning.yaml
euleragent graph compile examples/graphs/capstone/competitor_positioning.yaml \
  --out examples/graphs/capstone/competitor_positioning_compiled.json

# 잘못된 예: local_search는 HITL이 아닌데 approvals_resolved 사용
- from: local_search
  to: merge_findings
  when: "approvals_resolved"  # ← local_search는 execute mode! 항상 true여야 함

# OK: doc_search는 findings만 씀, source_count는 안 씀
- id: doc_search
  writes_state: [findings]   # source_count 없음 — 허용

자동 동작:
  web_search (HITL 대기 중)
  local_search 완료 → merge_findings 대기
  doc_search 완료 → merge_findings 대기
  web_search 승인 완료 → merge_findings 실행! (세 브랜치 모두 완료)