Orchestration Engine

The Orchestration Engine is the core agent execution runtime for AEGIS. It uses LangGraph to build a stateful graph pipeline that processes user messages through system prompts, memory retrieval, LLM calls, tool execution, skill injection, HITL approval checkpoints, and output formatting.

Overview

This service is the brain of AEGIS. When a user sends a message to an AI agent, the orchestration engine:

1. Loads the agent’s system prompt and conversation history
2. Retrieves working memory and episodic memories
3. Calls the LLM (via LiteLLM) with budget enforcement
4. Executes any tool calls the LLM requests
5. Detects and injects relevant skills (3-tier architecture)
6. Checks if HITL approval is required
7. Runs a synthesis LLM call with injected skill context
8. Formats the final output

Beyond the core agent pipeline, the orchestration engine also hosts the compliance dashboard API, checklist management, rule lifecycle, and workspace SSE endpoints.

Port & Language

Key Endpoints

Agent Execution

Conversations

Compliance Dashboard (R1.2)

Checklists (R1.3)

Artifacts

Rules (R1.5)

Workspaces (R1.4)

Architecture

LangGraph Pipeline

The core execution follows a StateGraph with 9 nodes:

START -> system_prompt_node -> memory_node -> initial_llm_call
      -> [tool_node -> initial_llm_call]* (tool loop)
      -> skill_select_node -> [skill_inject_node] -> approval_node
      -> synthesis_llm_call -> [tool_node -> synthesis_llm_call]*
      -> output_format -> END

Routing logic between nodes:

• After initial_llm_call: If there are pending tool calls, route to tool_node. If there is an error, route to output_format. Otherwise, route to skill_select_node.
• After tool_node: Check iteration limit. Route back to the appropriate LLM call based on the current phase (initial or synthesis).
• After skill_select_node: If skills were detected, route to skill_inject_node. Otherwise, skip to approval_node.
• After approval_node: If HITL is required, route to output_format with awaiting_hitl status. Otherwise, continue to synthesis_llm_call.
• After synthesis_llm_call: If tool calls are needed, route to tool_node. Otherwise, route to output_format.

Module Breakdown

src/orchestration/
├── main.py                     # FastAPI app, /execute and /stream endpoints
├── engine.py                   # LangGraph StateGraph construction and compilation
├── nodes.py                    # Node implementations (1 function per node)
├── tools.py                    # Tool definitions and execution for LLM function calling
├── state.py                    # GraphState TypedDict (flows through the graph)
├── schemas.py                  # Request/response Pydantic models
├── services.py                 # HTTP clients for memory, KG, approval services
├── budget.py                   # Token/cost budget enforcement
├── config.py                   # Settings from environment variables
├── conversation_routes.py      # Conversation CRUD (list, create, get, update, delete)
├── skills/                     # Checklist-driven skill implementations
│   ├── __init__.py             # Skill registry (register_skill, list_registered_skills)
│   ├── base.py                 # Base skill class
│   ├── rule37/                 # Rule 37 spacing exception skills (11 skills)
│   ├── rule32/                 # Rule 32 flaring exception skills (10 skills)
│   ├── form_pr/                # Form PR production report skills (8 skills)
│   ├── flaring_monitor/        # Flaring monitor skills (6 skills)
│   └── field_event/            # Field event conversation skills
├── compliance/                 # Compliance dashboard and workspace
│   ├── routes.py               # /compliance/summary, /matrix, /charts
│   ├── engine.py               # Compliance status computation engine
│   ├── queries.py              # Cypher query templates for compliance
│   ├── checklist_routes.py     # Checklist CRUD routes
│   ├── artifact_routes.py      # Artifact management routes
│   ├── rule_routes.py          # Rule version CRUD routes
│   ├── rule_lifecycle.py       # Immutable rule versioning logic
│   ├── rule_monitor/           # Rule change monitoring
│   └── workspace/              # Entity workspace (assessment, SSE, agent)
│       ├── routes.py           # Workspace API routes
│       ├── assessment.py       # Automated assessment logic
│       └── events.py           # WorkspaceEventType enum, SSE event formatting
├── seed_skills.py              # Skill registry seeding script
├── seed_checklists.py          # Checklist template seeding script
├── seed_rules.py               # Compliance rule data seeding script
├── seed_conversations.py       # Demo conversation seeding
└── seed_demo_data.py           # Combined demo data seeder

GraphState

The GraphState TypedDict carries all execution state through the graph:

class GraphState(TypedDict, total=False):
    execution_id: str
    conversation_id: str
    agent_id: str
    agent_type: str              # "rule_37", "rule_32", etc.
    tenant_id: str
    status: str                  # "pending", "running", "completed", "awaiting_hitl", "failed"
    messages: list[dict]         # Conversation messages (role/content/tool_calls)
    system_prompt: str
    model: str                   # LLM model to use (default: gpt-4o)
    injected_skill_ids: list     # Skills already injected
    active_skills: dict          # skill_id -> tier level
    tokens_used: int
    cost_usd: float
    max_tokens: int
    max_cost_usd: float
    pending_tool_calls: list     # Tool calls awaiting execution
    iteration: int               # Guards against infinite loops
    memory_context: dict         # Working + episodic memory data
    selected_skill_ids: list     # Skills detected from LLM output
    llm_phase: str               # "initial" or "synthesis"
    hitl_required: bool
    hitl_checkpoint_type: str
    hitl_approval_id: str | None

Budget Enforcement

The budget.py module enforces per-execution limits:

• Token budget: Raises BudgetExceededError when tokens_used >= max_tokens
• Cost budget: Raises BudgetExceededError when cost_usd >= max_cost_usd

Budget is checked before every LLM call. When exceeded, the /execute endpoint returns HTTP 429. The SSE stream emits an error event.

Skill Injection (3-Tier Architecture)

When the LLM output contains SKILL_SELECT:{skill_id} patterns, the skill injection pipeline activates:

1. Ledger check — Query the memory service’s injection ledger to prevent duplicate injection
2. Tier 2 — Load the skill definition (description, steps, requirements, output format) from the PostgreSQL skills table
3. Tier 3 — Load reference artifacts (templates, examples, regulations) from the skill_artifacts table
4. Tier 3.5 — Load entity context from the knowledge graph (well data, lease data, relationships)
5. Mark ledger — Record the injection to prevent re-injection in subsequent turns

All injected content is appended as system messages to the conversation.

HITL Approval Checkpoints

The approval node auto-detects HITL requirements from:

• Injected skills with known checkpoint mappings (e.g., rule37-filing-assembly triggers pre_filing)
• Executed tools (e.g., rule37_filing_assembly triggers pre_filing)
• Metadata overrides (hitl_checkpoint_type in request metadata)

When HITL is required, the engine creates an approval request via the approval service and pauses execution with status: "awaiting_hitl".

Tool Definitions

The engine registers tools for LLM function calling based on agent type:

Dependencies

Python Packages

Service Dependencies

Configuration

Observability (Langfuse Tracing)

The engine ships optional Langfuse  tracing, wired in orchestration/observability.py. It is off by default and activates only when both LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY are set — otherwise every tracing helper is a no-op and the service behaves identically.

How it works:

• At startup (init_tracing() in the FastAPI lifespan) the engine enables the LiteLLM langfuse_otel callback for all litellm.acompletion calls and initialises the Langfuse SDK (v4, OpenTelemetry-based).
• Each agent run is wrapped in a parent span via trace_agent_run(...). LiteLLM generation spans nest under it automatically: both layers ride the global OTEL TracerProvider and share OTEL context, and Langfuse’s span processor exports gen_ai.* spans by default — yielding one trace per agent run with each LLM call as a nested generation (model, tokens and cost captured automatically).

Trace attributes set:

Instrumented entry points: POST /execute, GET /conversations/{id}/stream, workspace assessment (GET /workspaces/{id}/stream), and the scoped agent (POST /workspaces/{id}/items/{idx}/agent).

R34 entity-layer child spans (orchestration/entity_spans.py): the entity resolution / context-assembly / HITL pipelines emit child_span observations that nest under the agent-run span, so an auto-select decision and the full HITL flow are reconstructable from the trace alone. They carry the internal, precise values (the audit audience, not the LLM) and are no-ops when tracing is off. All fire orchestration-side — the knowledge-graph service is untraced and OTEL context doesn’t cross the HTTP boundary, so KG returns the facts (the resolution_trace block on the resolve response) and the orchestration tool wrappers + gate/guard emit the spans.

The entity_resolve.outcome span is a separate layer from the platform:log_entity_resolution rules-audit entry — the spans only wrap the resolve callsite and never re-fire the on_entity_resolve trigger (one audit entry + one outcome span per auto-select). An approved-but-unexecuted mutation is the absence of a hitl.resume_replay span next to an approved approval record. See docs/specs/phases/R34-phase-6-calibration-and-demo-runbook.md.

Running Locally

cd services/orchestration-engine
poetry install
poetry run uvicorn orchestration.main:app --reload --port 8001

Seeding Data

The service includes several seeding scripts:

# Seed the skill registry and agent definitions
poetry run python -m orchestration.seed_skills
 
# Seed checklist templates
poetry run python -m orchestration.seed_checklists
 
# Seed compliance rules
poetry run python -m orchestration.seed_rules
 
# Seed demo conversations
poetry run python -m orchestration.seed_conversations