노드, 간선 및 상태

State 객체

from typing import TypedDict
from typing_extensions import Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]   # appends instead of replaces
    plan: str
    answer: str

State 업데이트 작동 방식

Defining a Node

def planner(state: AgentState) -> dict:
    prompt = f'User wants: {state["messages"][-1].content}'
    plan = llm.invoke(prompt).content
    return {'plan': plan}

Registering the Node

from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)
graph.add_node('planner', planner)

간선 정의하기

graph.set_entry_point('planner')
graph.add_edge('planner', 'researcher')
graph.add_edge('researcher', 'writer')
graph.add_edge('writer', END)

조건부 간선

def route(state: AgentState) -> str:
    if state['needs_research']:
        return 'researcher'
    return 'writer'

graph.add_conditional_edges(
    'planner',
    route,
    {'researcher': 'researcher', 'writer': 'writer'}
)

END는 특별합니다

Compiling the Graph

app = graph.compile()
result = app.invoke({'messages': [...]})
print(result['answer'])

상태 업데이트 스트리밍

for event in app.stream({'messages': [...]}):
    print(event)
# Each event is the state diff from the just-completed node.

리듀서 함수

def merge_facts(old, new):
    return {**old, **new}

class State(TypedDict):
    facts: Annotated[dict, merge_facts]

하위 그래프

subgraph = StateGraph(SubState).compile()
graph.add_node('subprocess', subgraph)

비동기 노드

async def fetch_node(state):
    data = await fetch_async(state['url'])
    return {'data': data}

result = await app.ainvoke({...})

노드의 오류

def safe_search(state):
    try:
        return {'results': search(state['query'])}
    except Exception as e:
        return {'error': str(e), 'results': []}

State 업데이트

복습