Core Concepts - Upsonic AI

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:

Reducer	Behavior	Use Case
`operator.add`	Append/concatenate	Lists, strings
`lambda a, b: a + b`	Sum values	Counters, totals
`max`	Keep maximum	Scores, priorities
`min`	Keep minimum	Costs, 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:

Dictionary - State updates (merged with reducers)
Command - State updates + routing
Send - Dynamic parallel invocation
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

Quick Start - Build your first graph
Attributes - Comprehensive configuration guide
Advanced Features - Send API and task decorators
Building Agents - Create AI agents with tools

​States

​Basic State Definition

​State Reducers

​Example: Message History

​Nodes

​Basic Node

​Node Signatures

​Node Return Types

​Edges

​Simple Edges

​Conditional Edges

​Conditional Routing

​Method 1: Conditional Edges

​Method 2: Command Objects

​Loops and Cycles

​Best Practices

​1. Keep Nodes Focused

​2. Use Type Hints

​3. Design State Carefully

​Next Steps

States

Basic State Definition

State Reducers

Example: Message History

Nodes

Basic Node

Node Signatures

Node Return Types

Edges

Simple Edges

Conditional Edges

Conditional Routing

Method 1: Conditional Edges

Method 2: Command Objects

Loops and Cycles

Best Practices

1. Keep Nodes Focused

2. Use Type Hints

3. Design State Carefully

Next Steps