from ai_infra.graph import GraphProduction-ready workflow graph with zero-config building. Supports multiple initialization patterns: 1. Simple dict-based (GOAL: minimal boilerplate):
graph = Graph(
nodes={
"analyze": analyze_func,
"decide": decide_func,
"execute": execute_func,
},
edges=[
("analyze", "decide"),
("decide", "execute", lambda state: state["should_execute"]),
],
entry="analyze",
)2. Any callable as node:
graph = Graph(nodes={
"func": my_function,
"lambda": lambda state: {"output": state["input"] * 2},
"method": my_instance.method,
"agent": my_agent.run,
})3. Full LangGraph power:
graph = Graph(
nodes={...},
edges=[...],
checkpointer=MemorySaver(),
interrupt_before=["dangerous_node"],
interrupt_after=["checkpoint_node"],
)4. Type-safe state:
class WorkflowState(TypedDict):
input: str
output: str
graph = Graph[WorkflowState](
nodes={...},
state_schema=WorkflowState,
validate_state=True,
)