106. LangGraph: 상태 기반 에이전트 워크플로우

LangChain 체인은 한 방향으로 흐릅니다. 입력이 들어오면 출력이 나가고, 끝입니다.

실제 에이전트 워크플로우는 선형적이지 않습니다. 계획을 수정해야 할 수도 있고, 검색 결과가 없어서 다른 접근법을 시도해야 할 수도 있습니다. 코드 리뷰가 실패하면 작업을 다시 수정하도록 되돌려야 하고, 작업은 받는 입력 종류에 따라 다른 경로로 분기될 수 있습니다.

LangGraph는 에이전트 워크플로우를 방향 그래프로 모델링합니다. 노드는 행동을, 간선은 조건부 전이를 나타냅니다. 에이전트의 상태는 그래프를 따라 흐르면서 각 단계에서 발견한 내용에 따라 서로 다른 경로를 택합니다.

결과: 워크플로우를 언제든지 검사·디버깅·재개할 수 있고, 복잡한 조건 로직을 스파게티 코드가 되지 않게 구현할 수 있습니다.

---

LangGraph가 LangChain에 추가하는 것

print("LangGraph: What It Adds")
print()
print("LangChain gives you: components (LLMs, retrievers, tools)")
print("LangGraph gives you: orchestration as a directed graph")
print()

differences = {
    "Control flow":   ("Linear chain",        "Conditional graph with loops and branches"),
    "State":          ("Passed between steps", "Typed state shared across all nodes"),
    "Debugging":      ("Trace the chain",      "Visualize the graph, step through nodes"),
    "Resume":         ("Start over",           "Checkpoint any node, resume from there"),
    "Parallelism":    ("Sequential only",      "Parallel branches in the graph"),
    "Human-in-loop":  ("Not supported",        "Pause graph, wait for human, resume"),
}

print(f"{'Feature': dict:")
print("       # do work, return updates")
print("       return {'field': new_value}")
print()
print("3. EDGES: Connections between nodes")
print("   - Unconditional: always go to next node")
print("   - Conditional: function decides which node comes next")
print()
print("4. COMPILE: StateGraph().compile() turns the graph into a runnable")

---

최소 예시: Intent Router

class RouterState(TypedDict):
    user_input:   str
    intent:       str
    response:     str

def detect_intent(state: RouterState) -> dict:
    """Node 1: Classify the user's intent."""
    prompt  = f"""Classify this input into exactly one category:
    - math: involves calculation or numbers
    - code: involves programming
    - general: anything else

    Input: {state['user_input']}
    Respond with only the category name."""

    intent = (llm | parser).invoke(prompt).strip().lower()
    if intent not in ["math", "code", "general"]:
        intent = "general"

    print(f"  [detect_intent] → '{intent}'")
    return {"intent": intent}

def handle_math(state: RouterState) -> dict:
    """Node 2a: Handle math queries."""
    response = (llm | parser).invoke(
        f"Solve this math problem clearly: {state['user_input']}")
    print(f"  [handle_math] → answered")
    return {"response": response}

def handle_code(state: RouterState) -> dict:
    """Node 2b: Handle code queries."""
    response = (llm | parser).invoke(
        f"Answer this coding question with example code: {state['user_input']}")
    print(f"  [handle_code] → answered")
    return {"response": response}

def handle_general(state: RouterState) -> dict:
    """Node 2c: Handle general queries."""
    response = (llm | parser).invoke(state["user_input"])
    print(f"  [handle_general] → answered")
    return {"response": response}

def route_by_intent(state: RouterState) -> Literal["handle_math", "handle_code", "handle_general"]:
    """Conditional edge: decide which handler to call."""
    return f"handle_{state['intent']}" if state["intent"] in ["math", "code"] else "handle_general"

graph = StateGraph(RouterState)
graph.add_node("detect_intent", detect_intent)
graph.add_node("handle_math",    handle_math)
graph.add_node("handle_code",    handle_code)
graph.add_node("handle_general", handle_general)

graph.add_edge(START, "detect_intent")
graph.add_conditional_edges("detect_intent", route_by_intent)
graph.add_edge("handle_math",    END)
graph.add_edge("handle_code",    END)
graph.add_edge("handle_general", END)

router_app = graph.compile()

print("Intent Router Graph:")
test_inputs = [
    "What is 15% of 840?",
    "How do I reverse a list in Python?",
    "What is the capital of Japan?",
]
for inp in test_inputs:
    print(f"\nInput: '{inp}'")
    result = router_app.invoke({"user_input": inp, "intent": "", "response": ""})
    print(f"Response: {result['response'][:120]}...")

---

루프가 포함된 다단계 에이전트

class ResearchState(TypedDict):
    topic:         str
    search_queries: List[str]
    findings:      List[str]
    draft:         str
    review_score:  int
    final_output:  str
    iteration:     int

def plan_research(state: ResearchState) -> dict:
    """Generate search queries for the topic."""
    response = (llm | parser).invoke(
        f"Generate 3 specific search queries to research: {state['topic']}\n"
        f"Output as a numbered list, one query per line.")
    queries = [line.strip().lstrip("123. ").strip()
               for line in response.split("\n")
               if line.strip() and line.strip()[0].isdigit()][:3]
    print(f"  [plan] Generated {len(queries)} queries")
    return {"search_queries": queries}

def execute_search(state: ResearchState) -> dict:
    """Simulate searching and gathering findings."""
    findings = []
    for i, query in enumerate(state["search_queries"], 1):
        finding = (llm | parser).invoke(
            f"Provide 2-3 key facts about: {query}\nBe specific and concise.")
        findings.append(f"Query {i}: {query}\n{finding}")
        print(f"  [search {i}] gathered findings")
    return {"findings": findings}

def write_draft(state: ResearchState) -> dict:
    """Write a draft based on findings."""
    findings_text = "\n\n".join(state["findings"])
    draft = (llm | parser).invoke(
        f"Write a 3-paragraph summary about '{state['topic']}' "
        f"based on these findings:\n\n{findings_text}")
    print(f"  [write] draft written ({len(draft.split())} words)")
    return {"draft": draft, "iteration": state.get("iteration", 0) + 1}

def review_draft(state: ResearchState) -> dict:
    """Score the draft quality."""
    response = (llm | parser).invoke(
        f"Rate this text quality 1-10 (just the number):\n\n{state['draft']}")
    import re
    score_match = re.search(r'\b([0-9]|10)\b', response)
    score = int(score_match.group()) if score_match else 7
    print(f"  [review] score: {score}/10 (iteration {state.get('iteration', 1)})")
    return {"review_score": score}

def revise_draft(state: ResearchState) -> dict:
    """Improve the draft based on review."""
    revised = (llm | parser).invoke(
        f"Improve this text. Make it clearer and more informative:\n\n{state['draft']}")
    print(f"  [revise] draft improved")
    return {"draft": revised}

def finalize(state: ResearchState) -> dict:
    """Package the final output."""
    print(f"  [finalize] approved after {state.get('iteration', 1)} iteration(s)")
    return {"final_output": state["draft"]}

def should_revise(state: ResearchState) -> Literal["revise_draft", "finalize"]:
    """Decide: revise or finalize based on score and iteration count."""
    iteration = state.get("iteration", 1)
    score     = state.get("review_score", 7)
    if score  dict:
    content = (llm | parser).invoke(
        "Write a 2-sentence marketing tagline for a machine learning platform.")
    print(f"  [generate] content ready for review")
    return {"content": content, "approved": False}

def human_review(state: ApprovalState) -> dict:
    """This node PAUSES the graph and waits for human input."""
    print(f"\n  [PAUSED] Human review required.")
    print(f"  Content: {state['content']}")
    print(f"  (In production: send notification, wait f