Spaces:
Running
A newer version of the Gradio SDK is available:
6.1.0
Graph Orchestration Architecture
Overview
DeepCritical implements a graph-based orchestration system for research workflows using Pydantic AI agents as nodes. This enables better parallel execution, conditional routing, and state management compared to simple agent chains.
Conversation History
DeepCritical supports multi-turn conversations through Pydantic AI's native message history format. The system maintains two types of history:
- User Conversation History: Multi-turn user interactions (from Gradio chat interface) stored as
list[ModelMessage] - Research Iteration History: Internal research process state (existing
Conversationmodel)
Message History Flow
Gradio Chat History β convert_gradio_to_message_history() β GraphOrchestrator.run(message_history)
β
GraphExecutionContext (stores message_history)
β
Agent Nodes (receive message_history via agent.run())
β
WorkflowState (persists user_message_history)
Usage
Message history is automatically converted from Gradio format and passed through the orchestrator:
# In app.py - automatic conversion
message_history = convert_gradio_to_message_history(history) if history else None
async for event in orchestrator.run(query, message_history=message_history):
yield event
Agents receive message history through their run() methods:
# In agent execution
if message_history:
result = await agent.run(input_data, message_history=message_history)
Graph Patterns
Iterative Research Graph
The iterative research graph follows this pattern:
[Input] β [Thinking] β [Knowledge Gap] β [Decision: Complete?]
β No β Yes
[Tool Selector] [Writer]
β
[Execute Tools] β [Loop Back]
Node IDs: thinking β knowledge_gap β continue_decision β tool_selector/writer β execute_tools β (loop back to thinking)
Special Node Handling:
execute_tools: State node that usessearch_handlerto execute searches and add evidence to workflow statecontinue_decision: Decision node that routes based onresearch_completeflag fromKnowledgeGapOutput
Deep Research Graph
The deep research graph follows this pattern:
[Input] β [Planner] β [Store Plan] β [Parallel Loops] β [Collect Drafts] β [Synthesizer]
β β β
[Loop1] [Loop2] [Loop3]
Node IDs: planner β store_plan β parallel_loops β collect_drafts β synthesizer
Special Node Handling:
planner: Agent node that createsReportPlanwith report outlinestore_plan: State node that storesReportPlanin context for parallel loopsparallel_loops: Parallel node that executesIterativeResearchFlowinstances for each sectioncollect_drafts: State node that collects section drafts from parallel loopssynthesizer: Agent node that callsLongWriterAgent.write_report()directly withReportDraft
Deep Research
sequenceDiagram
actor User
participant GraphOrchestrator
participant InputParser
participant GraphBuilder
participant GraphExecutor
participant Agent
participant BudgetTracker
participant WorkflowState
User->>GraphOrchestrator: run(query)
GraphOrchestrator->>InputParser: detect_research_mode(query)
InputParser-->>GraphOrchestrator: mode (iterative/deep)
GraphOrchestrator->>GraphBuilder: build_graph(mode)
GraphBuilder-->>GraphOrchestrator: ResearchGraph
GraphOrchestrator->>WorkflowState: init_workflow_state()
GraphOrchestrator->>BudgetTracker: create_budget()
GraphOrchestrator->>GraphExecutor: _execute_graph(graph)
loop For each node in graph
GraphExecutor->>Agent: execute_node(agent_node)
Agent->>Agent: process_input
Agent-->>GraphExecutor: result
GraphExecutor->>WorkflowState: update_state(result)
GraphExecutor->>BudgetTracker: add_tokens(used)
GraphExecutor->>BudgetTracker: check_budget()
alt Budget exceeded
GraphExecutor->>GraphOrchestrator: emit(error_event)
else Continue
GraphExecutor->>GraphOrchestrator: emit(progress_event)
end
end
GraphOrchestrator->>User: AsyncGenerator[AgentEvent]
Iterative Research
sequenceDiagram
participant IterativeFlow
participant ThinkingAgent
participant KnowledgeGapAgent
participant ToolSelector
participant ToolExecutor
participant JudgeHandler
participant WriterAgent
IterativeFlow->>IterativeFlow: run(query)
loop Until complete or max_iterations
IterativeFlow->>ThinkingAgent: generate_observations()
ThinkingAgent-->>IterativeFlow: observations
IterativeFlow->>KnowledgeGapAgent: evaluate_gaps()
KnowledgeGapAgent-->>IterativeFlow: KnowledgeGapOutput
alt Research complete
IterativeFlow->>WriterAgent: create_final_report()
WriterAgent-->>IterativeFlow: final_report
else Gaps remain
IterativeFlow->>ToolSelector: select_agents(gap)
ToolSelector-->>IterativeFlow: AgentSelectionPlan
IterativeFlow->>ToolExecutor: execute_tool_tasks()
ToolExecutor-->>IterativeFlow: ToolAgentOutput[]
IterativeFlow->>JudgeHandler: assess_evidence()
JudgeHandler-->>IterativeFlow: should_continue
end
end
Graph Structure
Nodes
Graph nodes represent different stages in the research workflow:
Agent Nodes: Execute Pydantic AI agents
- Input: Prompt/query
- Output: Structured or unstructured response
- Examples:
KnowledgeGapAgent,ToolSelectorAgent,ThinkingAgent
State Nodes: Update or read workflow state
- Input: Current state
- Output: Updated state
- Examples: Update evidence, update conversation history
Decision Nodes: Make routing decisions based on conditions
- Input: Current state/results
- Output: Next node ID
- Examples: Continue research vs. complete research
Parallel Nodes: Execute multiple nodes concurrently
- Input: List of node IDs
- Output: Aggregated results
- Examples: Parallel iterative research loops
Edges
Edges define transitions between nodes:
Sequential Edges: Always traversed (no condition)
- From: Source node
- To: Target node
- Condition: None (always True)
Conditional Edges: Traversed based on condition
- From: Source node
- To: Target node
- Condition: Callable that returns bool
- Example: If research complete β go to writer, else β continue loop
Parallel Edges: Used for parallel execution branches
- From: Parallel node
- To: Multiple target nodes
- Execution: All targets run concurrently
State Management
State is managed via WorkflowState using ContextVar for thread-safe isolation:
- Evidence: Collected evidence from searches
- Conversation: Iteration history (gaps, tool calls, findings, thoughts)
- Embedding Service: For semantic search
State transitions occur at state nodes, which update the global workflow state.
Execution Flow
- Graph Construction: Build graph from nodes and edges using
create_iterative_graph()orcreate_deep_graph() - Graph Validation: Ensure graph is valid (no cycles, all nodes reachable) via
ResearchGraph.validate_structure() - Graph Execution: Traverse graph from entry node using
GraphOrchestrator._execute_graph() - Node Execution: Execute each node based on type:
- Agent Nodes: Call
agent.run()with transformed input - State Nodes: Update workflow state via
state_updaterfunction - Decision Nodes: Evaluate
decision_functionto get next node ID - Parallel Nodes: Execute all parallel nodes concurrently via
asyncio.gather()
- Agent Nodes: Call
- Edge Evaluation: Determine next node(s) based on edges and conditions
- Parallel Execution: Use
asyncio.gather()for parallel nodes - State Updates: Update state at state nodes via
GraphExecutionContext.update_state() - Event Streaming: Yield
AgentEventobjects during execution for UI
GraphExecutionContext
The GraphExecutionContext class manages execution state during graph traversal:
- State: Current
WorkflowStateinstance - Budget Tracker:
BudgetTrackerinstance for budget enforcement - Node Results: Dictionary storing results from each node execution
- Visited Nodes: Set of node IDs that have been executed
- Current Node: ID of the node currently being executed
Methods:
set_node_result(node_id, result): Store result from node executionget_node_result(node_id): Retrieve stored resulthas_visited(node_id): Check if node was visitedmark_visited(node_id): Mark node as visitedupdate_state(updater, data): Update workflow state
Conditional Routing
Decision nodes evaluate conditions and return next node IDs:
- Knowledge Gap Decision: If
research_completeβ writer, else β tool selector - Budget Decision: If budget exceeded β exit, else β continue
- Iteration Decision: If max iterations β exit, else β continue
Parallel Execution
Parallel nodes execute multiple nodes concurrently:
- Each parallel branch runs independently
- Results are aggregated after all branches complete
- State is synchronized after parallel execution
- Errors in one branch don't stop other branches
Budget Enforcement
Budget constraints are enforced at decision nodes:
- Token Budget: Track LLM token usage
- Time Budget: Track elapsed time
- Iteration Budget: Track iteration count
If any budget is exceeded, execution routes to exit node.
Error Handling
Errors are handled at multiple levels:
- Node Level: Catch errors in individual node execution
- Graph Level: Handle errors during graph traversal
- State Level: Rollback state changes on error
Errors are logged and yield error events for UI.
Backward Compatibility
Graph execution is optional via feature flag:
USE_GRAPH_EXECUTION=true: Use graph-based executionUSE_GRAPH_EXECUTION=false: Use agent chain execution (existing)
This allows gradual migration and fallback if needed.
See Also
- Orchestrators - Overview of all orchestrator patterns
- Workflow Diagrams - Detailed workflow diagrams
- API Reference - Orchestrators - API documentation