그래프 01. Graph vs Pattern — 무엇이 다른가?

# 환경 확인
euleragent --version
pip show langgraph | grep Version
euleragent graph --help

Pattern YAML ──────────────────────────────────────────
  ┌─────────────────────────────────────────────────┐
  │  id, version, category, description             │
  │  defaults (max_iterations, ...)                 │
  │  nodes (llm, judge, finalize kinds)             │
  │  edges (from, to, when conditions)              │
  │  finalize                                       │
  └─────────────────────────────────────────────────┘

Graph YAML ─────────────────────────────────────────────
  ┌─────────────────────────────────────────────────┐
  │  [Pattern의 모든 필드 동일하게 사용 가능]        │
  │                                                 │
  │  + state_schema      ← 신규 (타입+리듀서 정의)  │
  │  + parallel_groups   ← 신규 (팬아웃/팬인)       │
  │  + interrupt_before  ← 신규 (노드 전 일시 정지) │
  │  + interrupt_after   ← 신규 (노드 후 일시 정지) │
  └─────────────────────────────────────────────────┘

Pattern 실행 흐름:
  YAML → euleragent 순차 실행 엔진 → 결과

Graph 실행 흐름:
  YAML → IR 컴파일 → LangGraph StateGraph → 결과

state_schema:
  findings:          # 상태 키 이름
    type: list       # 데이터 타입
    merge: append_list   # 리듀서 전략
  score:
    type: integer
    merge: sum_int
  summary:
    type: string
    merge: last_write

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

checkpointer: memory   # ← interrupt 사용 시 최상위에 반드시 선언 (INTERRUPT_REQUIRES_CHECKPOINTER)

nodes:
  - id: publish
    kind: llm
    interrupt_before: true   # publish 실행 전 일시 정지
    runner:
      mode: execute

# LangGraph가 내부적으로 생성하는 상태 타입 (개념적 표현)
from typing import TypedDict, Annotated
import operator

class GraphState(TypedDict):
    findings: Annotated[list, operator.add]     # append_list → operator.add
    score: Annotated[int, operator.add]         # sum_int → operator.add
    summary: str                                # last_write → 기본값

잘못된 이해:
  브랜치 A의 상태 ──→ (격리됨) ──→ 독립 상태
  브랜치 B의 상태 ──→ (격리됨) ──→ 독립 상태

올바른 이해:
  브랜치 A ──→ 공유 상태 딕셔너리 ←── 브랜치 B
                     ↕
  리듀서가 충돌을 해결함 (operator.add 등)

# euleragent가 내부적으로 생성하는 LangGraph 코드 (개념적 표현)
def route_judge(state: GraphState) -> str:
    return state.get("route", "finalize")

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

┌─────────────────────────────────────────────────────────────────┐
│ 단계 1: YAML 파싱                                               │
│   your_graph.yaml                                               │
│        ↓ (PyYAML 파싱)                                          │
│   Python 딕셔너리                                               │
└─────────────────────────────────────────────────────────────────┘
         ↓
┌─────────────────────────────────────────────────────────────────┐
│ 단계 2: 검증 (euleragent graph validate)                        │
│   - Pattern 기본 검증 (노드 존재, 엣지 연결 등)                 │
│   - Graph 추가 검증 (병렬 에러 코드 14개 확인)                  │
│   - state_schema 타입+merge 호환성 검사                         │
└─────────────────────────────────────────────────────────────────┘
         ↓
┌─────────────────────────────────────────────────────────────────┐
│ 단계 3: IR 생성 (euleragent graph compile)                      │
│   - Graph IR JSON 생성                                          │
│   - graph_type: "graph" 마킹                                    │
│   - LangGraph StateGraph 빌더 코드 포함                         │
│   - parallel_groups → LangGraph 병렬 엣지 변환                  │
│   - state_schema → Annotated 타입 변환                          │
└─────────────────────────────────────────────────────────────────┘
         ↓
┌─────────────────────────────────────────────────────────────────┐
│ 실행 시: LangGraph StateGraph 실행                              │
│   - IR을 LangGraph StateGraph로 구체화                          │
│   - .compile() 호출                                             │
│   - .invoke() / .stream() 실행                                  │
└─────────────────────────────────────────────────────────────────┘

euleragent doctor

euleragent doctor 실행 중...
  Python          3.11.4   ✓
  euleragent      0.9.0    ✓
  langgraph       1.0.9    ✓  (graph 모듈 사용 가능)
  pyyaml          6.0.1    ✓
  click           8.1.7    ✓

모든 진단 항목 통과.

pip install "langgraph>=1.0.9"

# 예제 디렉터리에 있는 병렬 그래프 검증
euleragent graph validate examples/graphs/parallel/02_research_parallel.yaml

검증 중: examples/graphs/parallel/02_research_parallel.yaml

단계 1/3: YAML 파싱... 완료
단계 2/3: Pattern 기본 검증...
  [✓] 노드 ID 유일성
  [✓] 엣지 소스/타겟 존재
  [✓] finalize 노드 도달 가능성
  [✓] judge route_values 커버리지
  완료
단계 3/3: Graph 추가 검증...
  [✓] state_schema 존재 (parallel_groups 사용)
  [✓] 모든 브랜치 writes_state 선언
  [✓] 조인 노드 존재 및 단일성
  [✓] 브랜치 수 제한 (≤8)
  [✓] 부작용 도구 미사용 (shell.exec, file.write)
  [✓] 모든 브랜치 조인 노드로 수렴
  [✓] state_schema merge 타입 호환성
  완료

결과: 유효 (오류 없음)

# 단순 선형 그래프 컴파일
euleragent graph compile examples/graphs/linear/01_simple_pipeline.yaml

{
  "graph_type": "graph",
  "id": "graph.simple_pipeline",
  "version": 1,
  "compiled_at": "2026-02-23T10:00:00Z",
  "langgraph_version": "1.0.9",
  "state_schema": null,
  "parallel_groups": [],
  "nodes": [
    {
      "id": "research",
      "kind": "llm",
      "runner": { "mode": "plan", "force_tool": "web.search" },
      "interrupt_before": false,
      "interrupt_after": false
    },
    {
      "id": "draft",
      "kind": "llm",
      "runner": { "mode": "execute" }
    },
    {
      "id": "finalize",
      "kind": "finalize"
    }
  ],
  "edges": [
    { "from": "research", "to": "draft", "when": "approvals_resolved" },
    { "from": "draft", "to": "finalize", "when": "true" }
  ],
  "langgraph_builder": {
    "add_nodes": ["research", "draft", "finalize"],
    "add_edges": [
      ["research", "draft"],
      ["draft", "finalize"]
    ],
    "add_conditional_edges": [],
    "interrupt_before": [],
    "interrupt_after": []
  }
}

euleragent graph compile examples/graphs/linear/01_simple_pipeline.yaml \
  --out /tmp/compiled_linear.json

# 내용 확인
python -m json.tool /tmp/compiled_linear.json | head -60

euleragent graph validate examples/graphs/parallel/02_research_parallel.yaml \
  --format json | python -m json.tool

{
  "valid": true,
  "errors": [],
  "warnings": [
    {
      "code": "PARALLEL_NONDETERMINISTIC_MERGE",
      "severity": "warning",
      "message": "노드 'summary_branch'가 last_write 키 'summary'에 씁니다. 2개 이상의 브랜치가 이 키를 쓰면 비결정론적 결과가 발생합니다.",
      "node": "summary_branch"
    }
  ],
  "graph_type": "graph",
  "parallel_groups_count": 1,
  "branch_count": 2
}

# 위험한 패턴 — state_schema 없이 parallel_groups 사용
parallel_groups:
  - id: grp
    branches: [branch_a, branch_b]
    join: join_node

nodes:
  - id: branch_a
    kind: llm
    writes_state: [findings]   # ← 'findings'에 쓰려고 함
  - id: branch_b
    kind: llm
    writes_state: [findings]   # ← 같은 키 'findings'에 쓰려고 함

# state_schema가 없으므로 LangGraph는 'findings'의 리듀서를 모름
# 런타임에 INVALID_CONCURRENT_GRAPH_UPDATE 발생!

# 올바른 패턴 — state_schema에 리듀서 명시
state_schema:
  findings:
    type: list
    merge: append_list   # ← 리듀서 선언으로 충돌 해결

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

# 에러 감지 예시
euleragent graph validate bad_example.yaml

# 출력:
# [오류] PARALLEL_STATE_SCHEMA_MISSING
#   parallel_groups가 선언되어 있지만 state_schema가 없습니다.
#   병렬 실행 시 INVALID_CONCURRENT_GRAPH_UPDATE 예외가 발생합니다.
#   해결: 최상위에 state_schema를 선언하고 각 브랜치의 writes_state
#   키에 대한 리듀서를 정의하세요.

# 개념적 Python 코드 — euleragent가 내부적으로 생성
import operator
from typing import Annotated, TypedDict

class GraphState(TypedDict):
    findings: Annotated[list, operator.add]   # append_list
    score: Annotated[int, operator.add]       # sum_int
    notes: Annotated[str, operator.add]       # concat_str
    summary: str                              # last_write (기본)

질문 1: 진정한 병렬 실행이 필요한가?
  → 예: 여러 독립적인 데이터 소스를 동시에 조회해야 함
  → 아니오: 순차 실행으로 충분함 → Pattern 사용

질문 2: 노드 수준 일시 정지(interrupt)가 필요한가?
  → 예: 특정 노드 전/후에 사람이 상태를 검사해야 함
  → 아니오: HITL force_tool로 충분 → Pattern 사용

질문 3: LangGraph의 체크포인팅/스트리밍이 필요한가?
  → 예: 그래프 상태를 중간에 저장하고 재개해야 함
  → 아니오: Pattern 사용

모두 아니오라면 → Pattern을 사용하십시오 (더 단순하고 안정적)
하나 이상 예라면 → Graph를 고려하십시오 (00_disclaimer.md 먼저 읽기)

# euleragent doctor
Python 3.11.4 ✓ | euleragent 0.9.0 ✓ | langgraph 1.0.9 ✓

# euleragent graph validate (유효한 경우)
결과: 유효 (오류 없음)

# euleragent graph compile
{
  "graph_type": "graph",
  "id": "...",
  "langgraph_builder": { ... }
}

오류: LANGGRAPH_COMPILE_FAILED
  langgraph 버전 0.2.x가 감지되었습니다. 1.0.9 이상이 필요합니다.
해결: pip install "langgraph>=1.0.9"

오류: No such command 'graph'.
해결: pip install -e . 로 재설치 후 euleragent graph --help 확인

오류: yaml.scanner.ScannerError: mapping values are not allowed here
해결: YAML 들여쓰기 확인 (스페이스 2칸 또는 4칸, 탭 혼용 금지)

오류: parallel_groups가 선언되어 있지만 state_schema가 없습니다.
해결: 최상위에 state_schema 섹션을 추가하고 각 브랜치의 writes_state
      키에 대한 type과 merge를 정의하세요.