Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.upsonic.ai/llms.txt

Use this file to discover all available pages before exploring further.

​
States

States are the data that flows through your graph. They’re defined as TypedDict for type safety.

​
Basic State Definition

from typing_extensions import TypedDict

class MyState(TypedDict):
    input: str
    output: str
    count: int

​
State Reducers

Reducers control how state updates are merged. By default, new values replace old ones: 
from typing import Annotated, List
import operator

class RichState(TypedDict):
    # Reducer: operator.add - new items append to list
    messages: Annotated[List[str], operator.add]
    
    # Reducer: sum - numeric values add together
    total_cost: Annotated[float, lambda a, b: a + b]
    
    # No reducer - new value replaces old (default)
    current_step: str
Common Reducer Patterns:
ReducerBehaviorUse Case
operator.addAppend/concatenateLists, strings
lambda a, b: a + bSum valuesCounters, totals
maxKeep maximumScores, priorities
minKeep minimumCosts, distances

​
Example: Message History

from typing import Annotated, List
import operator

class ChatState(TypedDict):
    messages: Annotated[List[str], operator.add]
    turn_count: Annotated[int, lambda a, b: a + b]
    topic: str

def node1(state: ChatState) -> dict:
    return {
        "messages": ["Hello"],  # Appends to list
        "turn_count": 1,        # Adds to counter
        "topic": "greeting"     # Replaces old value
    }

def node2(state: ChatState) -> dict:
    return {
        "messages": ["How are you?"],  # Appends
        "turn_count": 1                # Adds 1
    }

# After both nodes:
# messages = ["Hello", "How are you?"]
# turn_count = 2
# topic = "greeting"

​
Nodes

Nodes are functions that process state. They receive the current state and return updates.

​
Basic Node

def my_node(state: MyState) -> dict:
    """Process state and return updates."""
    result = process(state["input"])
    return {
        "output": result,
        "count": state["count"] + 1
    }

​
Node Signatures

Nodes can have different signatures: 
# 1. State only
def node(state: MyState) -> dict:
    return {"output": "done"}

# 2. State + config (access runtime context)
def node(state: MyState, config: dict) -> dict:
    context = config.get("context", {})
    return {"output": f"Using {context.get('model', 'default')}"}

# 3. Returning Command for explicit routing
from upsonic.graphv2 import Command, END

def node(state: MyState) -> Command:
    if state["count"] > 10:
        return Command(update={"status": "complete"}, goto=END)
    return Command(update={"count": state["count"] + 1}, goto="process")

​
Node Return Types

Nodes can return:
  1. Dictionary - State updates (merged with reducers)
  2. Command - State updates + routing
  3. Send - Dynamic parallel invocation
  4. List[Send] - Multiple parallel invocations
# Return dict
def simple_node(state) -> dict:
    return {"field": "value"}

# Return Command
def routing_node(state) -> Command:
    return Command(update={"field": "value"}, goto="next_node")

# Return Send (for parallel execution)
from upsonic.graphv2 import Send

def orchestrator(state) -> List[Send]:
    return [
        Send("worker", {"item": item})
        for item in state["items"]
    ]

​
Edges

Edges define the flow of execution between nodes.

​
Simple Edges

Direct connections from one node to another: 
from upsonic.graphv2 import START, END

builder.add_edge(START, "first_node")
builder.add_edge("first_node", "second_node")
builder.add_edge("second_node", END)

​
Conditional Edges

Branch based on state: 
def route_by_intent(state: MyState) -> str:
    """Return the name of the next node."""
    if state["intent"] == "question":
        return "answer"
    elif state["intent"] == "command":
        return "execute"
    else:
        return "fallback"

builder.add_conditional_edges(
    "classify",           # From node
    route_by_intent,      # Routing function
    ["answer", "execute", "fallback"]  # Possible targets
)

# Connect target nodes
builder.add_edge("answer", END)
builder.add_edge("execute", END)
builder.add_edge("fallback", END)
All nodes mentioned in targets must be connected to other nodes or END. The graph validator will catch missing connections.

​
Conditional Routing

​
Method 1: Conditional Edges

Use add_conditional_edges with a routing function: 
class State(TypedDict):
    score: int
    result: str

def route_by_score(state: State) -> str:
    if state["score"] >= 90:
        return "excellent"
    elif state["score"] >= 70:
        return "good"
    else:
        return "needs_improvement"

builder = StateGraph(State)
builder.add_node("evaluate", evaluate_node)
builder.add_node("excellent", excellent_handler)
builder.add_node("good", good_handler)
builder.add_node("needs_improvement", needs_improvement_handler)

builder.add_edge(START, "evaluate")
builder.add_conditional_edges(
    "evaluate",
    route_by_score,
    ["excellent", "good", "needs_improvement"]
)
builder.add_edge("excellent", END)
builder.add_edge("good", END)
builder.add_edge("needs_improvement", END)

​
Method 2: Command Objects

Nodes return Command to control routing: 
from upsonic.graphv2 import Command, END

def evaluate_node(state: State) -> Command:
    score = calculate_score(state)
    
    if score >= 90:
        return Command(
            update={"score": score, "result": "excellent"},
            goto="excellent"
        )
    elif score >= 70:
        return Command(
            update={"score": score, "result": "good"},
            goto="good"
        )
    else:
        return Command(
            update={"score": score, "result": "needs_improvement"},
            goto="needs_improvement"
        )

# With Command, you don't need add_conditional_edges
builder.add_edge(START, "evaluate")
builder.add_edge("excellent", END)
builder.add_edge("good", END)
builder.add_edge("needs_improvement", END)
Command vs Conditional Edges:
  • Use Command when the node itself decides routing (more explicit)
  • Use conditional_edges when routing logic should be separate from node logic

​
Loops and Cycles

Create loops by routing back to earlier nodes: 
from typing import TypedDict
from upsonic.graphv2 import Command, END, StateGraph, START

class LoopState(TypedDict):
    counter: int
    max_iterations: int
    result: str

def loop_node(state: LoopState) -> Command:
    new_counter = state["counter"] + 1
    
    if new_counter >= state["max_iterations"]:
        return Command(
            update={"counter": new_counter, "result": "done"},
            goto=END
        )
    else:
        return Command(
            update={"counter": new_counter},
            goto="loop_node"  # Loop back to self
        )

builder = StateGraph(LoopState)
builder.add_node("loop_node", loop_node)
builder.add_edge(START, "loop_node")

graph = builder.compile()

result = graph.invoke(
    {"counter": 0, "max_iterations": 5, "result": ""},
    config={"recursion_limit": 10}  # Prevent infinite loops
)

print(result["result"])  # "done"
print(result["counter"])  # 5
Recursion Limits: Always set a recursion_limit when using loops to prevent infinite execution. Default is 100 steps.

​
Best Practices

​
1. Keep Nodes Focused

Each node should have a single responsibility: 
# ✅ Good - focused nodes
def fetch_data(state): ...
def process_data(state): ...
def validate_results(state): ...

# ❌ Bad - doing too much
def big_node(state):
    # Fetch data
    # Process data
    # Validate
    # Store results
    pass

​
2. Use Type Hints

Always type your state and return values: 
# ✅ Good
def process(state: MyState) -> dict:
    return {"output": "result"}

​
3. Design State Carefully

Think about what needs to be in state vs what can be computed: 
# ✅ Good - only essential data
class State(TypedDict):
    items: List[str]
    # Compute count when needed: len(state["items"])

# ❌ Bad - storing derived data
class State(TypedDict):
    items: List[str]
    item_count: int  # Can be computed from items

​
Next Steps