Graph 10. Complete Reference — Fields, Error Codes, Safe Design Checklist

state_schema:
  <key_name>:
    type: string | integer | float | list | dict
    merge: append_list | sum_int | concat_str | last_write | first_write

parallel_groups:
  - id: <group_id>
    branches: [<node_id>, ...]
    join: <join_node_id>

# 병렬 결과 수집 (리스트) → append_list
findings:
  type: list
  merge: append_list

# 숫자 합산 → sum_int
count:
  type: integer
  merge: sum_int

# 텍스트 연결 (병렬) → concat_str
notes:
  type: string
  merge: concat_str

# 단일 노드 쓰기 → last_write
summary:
  type: string  # 또는 float, dict
  merge: last_write

# 최초 쓰기 우선 → first_write
initial_value:
  type: string  # 또는 다른 타입
  merge: first_write

□ 1. 진정한 병렬 실행이 필요한지 확인
      순차로 충분하면 Pattern 사용 (더 단순하고 안정적)

□ 2. 브랜치 수 결정 (2~8개)
      처음에는 2~3개로 시작하고 필요 시 확장

□ 3. state_schema 설계 완료
      모든 브랜치의 writes_state 키 목록화
      각 키의 type과 merge 전략 결정

□ 4. 공유 키(여러 브랜치가 씀) 식별
      공유 list 키 → append_list
      공유 integer 키 → sum_int
      공유 string 키 → concat_str 또는 last_write(비결정론적 허용 시)

□ 5. 단일 브랜치 키 식별
      단일 브랜치 키 → last_write (안전)

□ 6. state_schema 최상위에 선언

□ 7. parallel_groups 선언 (id, branches, join)

□ 8. 모든 브랜치 노드에 writes_state 선언
      상태를 쓰지 않아도 writes_state: [] 명시

□ 9. 조인 노드에 reads_state 선언 (문서화)

□ 10. 팬아웃 엣지 선언 (소스 → 모든 브랜치)

□ 11. 팬인 엣지 선언 (모든 브랜치 → 조인 노드)
       HITL 브랜치는 when: "approvals_resolved"
       자율 브랜치는 when: "true"

□ 12. interrupt_before/after 사용 시 최상위 checkpointer: 선언
       미선언 시 INTERRUPT_REQUIRES_CHECKPOINTER 에러 (정적 검증 실패)
       최소: checkpointer: memory

□ 13. euleragent graph validate 실행 및 통과 확인

□ 14. PARALLEL_NONDETERMINISTIC_MERGE / PARALLEL_ORDER_DEPENDENT_MERGE 경고 검토
       비결정론적 결과 허용 여부 확인; append_list 시 source/rank 필드 고려

□ 15. euleragent graph compile로 IR 생성 및 검토
       langgraph_builder 섹션에서 add_edges 확인

□ 16. 실험적 기능 사용 주의사항 재확인
       00_disclaimer.md 참조

from langgraph.graph import StateGraph, END
from typing import Annotated, TypedDict
import operator

# state_schema → TypedDict
class GraphState(TypedDict):
    findings: Annotated[list, operator.add]   # append_list
    source_count: Annotated[int, operator.add]  # sum_int
    final_summary: str                         # last_write (기본)
    route: str                                 # judge 라우팅용 내부 키

# StateGraph 초기화
graph = StateGraph(GraphState)

# 노드 추가
graph.add_node("plan", plan_fn)
graph.add_node("web_search", web_search_fn)
graph.add_node("local_search", local_search_fn)
graph.add_node("doc_search", doc_search_fn)
graph.add_node("merge_findings", merge_findings_fn)
graph.add_node("evaluate", evaluate_fn)
graph.add_node("revise", revise_fn)

# 진입점
graph.set_entry_point("plan")

# 팬아웃 엣지 (plan → 브랜치들)
graph.add_edge("plan", "web_search")
graph.add_edge("plan", "local_search")
graph.add_edge("plan", "doc_search")

# 팬인 엣지 (브랜치들 → 조인)
graph.add_edge("web_search", "merge_findings")
graph.add_edge("local_search", "merge_findings")
graph.add_edge("doc_search", "merge_findings")

# 하류 엣지
graph.add_edge("merge_findings", "evaluate")
graph.add_edge("revise", "evaluate")

# Judge 조건부 엣지
def route_evaluate(state: GraphState) -> str:
    return state.get("route", "finalize")

graph.add_conditional_edges(
    "evaluate",
    route_evaluate,
    {"finalize": END, "revise": "revise"}
)

# 컴파일 (interrupt 없는 경우)
compiled = graph.compile()

# 또는 interrupt 있는 경우
compiled = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["publish"],   # interrupt_before 노드 목록
    interrupt_after=["draft"]       # interrupt_after 노드 목록
)

def first_write_reducer(old_value, new_value):
    """첫 번째 쓰기 값을 유지. None이면 new_value 사용."""
    if old_value is None:
        return new_value
    return old_value  # 이미 값이 있으면 새 값 무시

# TypedDict에서의 선언
class GraphState(TypedDict):
    primary_result: Annotated[str, first_write_reducer]

# euleragent 또는 langgraph 업그레이드 후
pip install --upgrade euleragent langgraph

# 모든 Graph YAML 재검증
for yaml_file in examples/graphs/**/*.yaml; do
    echo "검증: $yaml_file"
    euleragent graph validate "$yaml_file"
done

# 컴파일된 IR 재생성
euleragent graph compile my_graph.yaml --out my_graph_compiled.json

□ euleragent doctor 실행 → 모든 항목 통과 확인
□ graph validate 통과 (오류 0개)
□ graph validate --format json 출력에서 경고 검토
□ graph compile 성공
□ 스테이징 환경에서 10회 이상 반복 실행 테스트
□ 병렬 결과의 비결정론적 변동 허용 범위 확인
□ HITL 승인 워크플로우 테스트
□ interrupt hooks 재개 시나리오 테스트
□ 감사 로그 (approvals.jsonl, tool_calls.jsonl) 검토
□ 최대 비용 시나리오 계산 (max_iterations × max_total_tool_calls)

euleragent graph list                              # 그래프 목록
euleragent graph show <path>                       # 구조 보기
euleragent graph validate <path>                   # 검증
euleragent graph validate <path> --format json     # JSON 형식 검증
euleragent graph compile <path>                    # IR 컴파일
euleragent graph compile <path> --out file.json    # IR 파일로 저장

id: graph.template
version: 1
category: demo
description: 최소 병렬 그래프 템플릿

state_schema:
  results:
    type: list
    merge: append_list

defaults:
  max_iterations: 2
  max_total_tool_calls: 30

parallel_groups:
  - id: my_group
    branches: [branch_a, branch_b]
    join: join_node

nodes:
  - id: source
    kind: llm
    runner: {mode: execute}

  - id: branch_a
    kind: llm
    runner: {mode: execute}
    writes_state: [results]

  - id: branch_b
    kind: llm
    runner: {mode: execute}
    writes_state: [results]

  - id: join_node
    kind: llm
    runner: {mode: execute}
    reads_state: [results]
    artifacts:
      primary: output.md

edges:
  - {from: source, to: branch_a, when: "true"}
  - {from: source, to: branch_b, when: "true"}
  - {from: branch_a, to: join_node, when: "true"}
  - {from: branch_b, to: join_node, when: "true"}
  - {from: join_node, to: finalize, when: "true"}

finalize:
  artifact: output.md