OSSS.ai.orchestration.graph_builder

OSSS.ai.orchestration.graph_builder

Graph builder for converting OSSS agents to LangGraph structures.

This module provides utilities to build LangGraph-compatible graphs from OSSS agent definitions, including: - node creation from agents - edge routing (dependencies + custom routing) - graph validation (missing nodes, cycles, entry/exit points) - a lightweight executor to simulate graph execution (sanity testing)

Important conceptual mapping: - A OSSS "agent" becomes a LangGraph "node" - Dependencies between agents become directed edges - The result is intended to be a DAG (Directed Acyclic Graph)

EdgeType

Bases: Enum

Types of edges in a LangGraph DAG.

This enum is used to label edges so downstream logic can interpret routing: - SEQUENTIAL: always traverse from A -> B after A completes - CONDITIONAL: traverse to B only if some condition holds (not fully evaluated yet) - PARALLEL: indicates branching can happen concurrently (not implemented in executor) - AGGREGATION: indicates many-to-one merge (not implemented in executor)

GraphEdge

Bases: BaseModel

Definition of an edge between graph nodes.

Why Pydantic? - Validates node IDs early (length, types, etc.) - Allows consistent serialization for logging/debugging - Fits the wider OSSS ecosystem which is Pydantic-first

NOTE about condition: - It is typed as Callable[[AgentContext], bool] - It is allowed as an arbitrary type (Pydantic config) - It is currently NOT evaluated by GraphExecutor (placeholder for future)

to_dict()

Convert to a simple dictionary representation.

NOTE: - We serialize edge_type as its .value (string) for readability. - We do NOT serialize the actual condition callable (non-serializable).

GraphDefinition

Bases: BaseModel

Complete graph definition consisting of nodes and edges.

This is the "compiled" representation produced by GraphBuilder.

Fields: - nodes: mapping of node_id -> LangGraphNodeDefinition (from each agent) - edges: list of GraphEdge objects describing connectivity - entry_points: nodes that can start execution (no incoming edges) - exit_points: nodes that can end execution (no outgoing edges) - metadata: extra descriptive/diagnostic info

to_dict()

Convert graph definition to a JSON-friendly dict.

NOTE: - node definitions are converted via node_def.to_dict() - edges are converted via edge.to_dict()

GraphValidationError

Bases: Exception

Error raised when a graph fails structural validation.

This exception is raised by :class:GraphBuilder when one of the following conditions is detected:

• No agents are provided.
• Edges reference nodes that do not exist in the graph.
• The graph contains cycles (it is not a DAG).
• Entry or exit points are missing or invalid.

GraphBuilder

Builder for creating LangGraph-compatible graphs from OSSS agents.

High-level flow: 1) Agents are registered with the builder. 2) Each agent provides a LangGraphNodeDefinition (node definition). 3) Edges are built from: - agent dependency declarations - optional custom edges - optional custom routing declarations 4) Entry/exit points are computed. 5) Graph is validated (missing nodes, cycles, etc.)

This does NOT create an actual LangGraph object yet. It produces a GraphDefinition that can be translated later.

add_agent(agent)

Register a single agent.

Implementation detail: - Stores the agent using agent.name.lower() as the key. - This makes subsequent node lookups consistent even if original names differ in case.

Returns:

Type	Description
GraphBuilder	Self (to enable fluent chaining)

add_agents(agents)

Register multiple agents.

This is convenience around add_agent().

Returns:

Type	Description
GraphBuilder	Self (to enable fluent chaining)

add_edge(edge)

Add an explicit edge.

Use cases: - Override default dependency wiring - Force ordering even if dependencies aren't declared - Add conditional/parallel/aggregation semantics

NOTE: - Validation still occurs in build()

add_conditional_routing(from_node, routing_func, condition_name='custom_routing')

Register custom routing behavior for a node.

Intended behavior (future): - After from_node executes, routing_func(context) decides which node to go next. - This could implement "retry", "branching", "quality gates", etc.

Current behavior: - GraphBuilder will generate CONDITIONAL edges from from_node to all other nodes. - GraphExecutor will currently take all conditional edges (placeholder).

Parameters:

Name	Type	Description	Default
from_node	str	the node that branches	required
routing_func	Callable[[AgentContext], str]	function mapping context -> next node name	required
condition_name	str	descriptive label (not used directly yet)	'custom_routing'

build()

Build and validate a GraphDefinition.

Steps: - Fail fast if no agents - Ask each agent for its node definition - Build edges (dependencies + custom edges + custom routing placeholders) - Determine entry and exit points - Validate the resulting graph (node existence, cycles, entry/exit correctness)

Raises:

Type	Description
GraphValidationError	If no agents are provided.
GraphValidationError	If an edge references a node that does not exist.
GraphValidationError	If the graph contains cycles.
GraphValidationError	If entry or exit points are missing or invalid.

GraphExecutor

Executor for running graphs built by GraphBuilder.

This is NOT a full LangGraph runtime. It is a lightweight simulator used to validate that: - Nodes can be executed in some order - Edges can be traversed - Context can flow through node invocations

Current limitations: - CONDITIONAL edges are not actually conditioned; all targets are taken. - PARALLEL edges are not executed concurrently. - AGGREGATION is not implemented. - Infinite-loop prevention is crude (execution_order length heuristic).

execute(initial_context) async

Execute the graph starting from entry points.

Algorithm (high level): - Start from graph_def.entry_points - For each node: - invoke the corresponding agent (if present) - add node to visited set - compute next nodes via outgoing edges - Continue until no next nodes remain or loop guard triggers

Returns:

Type	Description
AgentContext	The final AgentContext after executing reachable nodes.

Side effects

Adds execution trace metadata into context.execution_state: - graph_execution_order - graph_nodes_visited