Bedingtes Routing und Verzweigungen

Warum bedingte Edges?

Eine bedingte Edge hinzufügen

def route_after_planning(state: AgentState) -> str:
    if state.get('confidence', 0) > 0.8:
        return 'execute'
    return 'replan'

graph.add_conditional_edges(
    'planner',
    route_after_planning,
    {
        'execute': 'executor',
        'replan': 'planner'
    }
)

Signaturen von Routing-Funktionen

Selbstschleifen

graph.add_conditional_edges(
    'validator',
    lambda s: 'retry' if not s['valid'] else 'done',
    {'retry': 'planner', 'done': END}
)

Parallele Zweige

def fan_out(state):
    return ['search_news', 'search_arxiv', 'search_github']

graph.add_conditional_edges('planner', fan_out)

Parallele Zweige zusammenführen

Routing anhand von Tool-Aufrufen

def should_use_tool(state):
    last_msg = state['messages'][-1]
    if last_msg.tool_calls:
        return 'tools'
    return END

graph.add_conditional_edges('agent', should_use_tool, {'tools': 'tools', END: END})

Routing mit mehreren Optionen

def route_by_intent(state):
    intent = classify(state['messages'][-1])
    return {'q': 'qa_agent', 'order': 'order_agent', 'support': 'support_agent'}.get(intent, 'fallback')

Routing mit LLM-Aufrufen

def llm_router(state):
    decision = llm.invoke(f'Route this to: search, calculate, or end?\nState: {state}').content.strip()
    return decision  # 'search' / 'calculate' / END

Endlosschleifen vermeiden

class State(TypedDict):
    step: int
    ...

def route(state):
    if state['step'] >= MAX_STEPS:
        return END
    return 'continue'

Increment Counter Reducer

def add(old: int, new: int) -> int:
    return old + new

class State(TypedDict):
    step: Annotated[int, add]

def any_node(state):
    return {'step': 1}   # increments by 1

Zweige visualisieren

print(app.get_graph().draw_mermaid())

Bedingte Edges

Zusammenfassung