โหนด ขอบ และสถานะ

ออบเจ็กต์สถานะ

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

การทำงานของการอัปเดตสถานะ

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': []}

การอัปเดตสถานะ

สรุปทบทวน