Routing condizionale e branching

Perché usare gli archi condizionali?

Aggiunta di un arco condizionale

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'
    }
)

Firme delle funzioni di routing

Self-loop

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

Diramazioni parallele

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

graph.add_conditional_edges('planner', fan_out)

Unione delle diramazioni parallele

Routing in base alle chiamate agli strumenti

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 a più vie

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

Routing con chiamate LLM

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

Evita i loop infiniti

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

Visualizzazione delle diramazioni

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

Archi condizionali

Riepilogo