Orchestrator Hooks

The ContextGraphOrchestratorHooks class provides non-invasive middleware hooks that plug into both the BI Analytics Orchestrator and the General Agent Orchestrator. These hooks capture agent thinking steps, LLM calls, and tool executions as AgentThinkingStep records without modifying core orchestrator logic.

Overview

The hooks follow an event-driven middleware pattern: orchestrators call hook methods at key execution points, and the hooks delegate to the AgentThinkingCaptureService for persistence. All hook methods are safe to call even when the thinking service is unavailable (graceful degradation).

Source: data-plane/ai-service/src/context_graph/integration/orchestrator_hooks.py

Integration Points

BI Orchestrator

from context_graph.integration.orchestrator_hooks import get_orchestrator_hooks
 
hooks = get_orchestrator_hooks()
if hooks:
    trace_id = await hooks.on_bi_message_start(session_id, tenant_id, message)
    # ... run BI agents ...
    await hooks.on_bi_message_complete(trace_id, response)

General Orchestrator

hooks = get_orchestrator_hooks()
if hooks:
    trace_id = await hooks.on_agent_start(session_id, tenant_id, goal)
    await hooks.on_agent_step(trace_id, step_type, reasoning, confidence)
    await hooks.on_agent_complete(trace_id, outcome)

Hook Methods

Method	When Called	What It Captures
`on_bi_message_start`	BI agent begins processing a user message	Creates a new thinking trace
`on_bi_message_complete`	BI agent finishes processing	Completes the trace with outcome and metrics
`on_agent_start`	General agent begins a task	Creates a new thinking trace
`on_agent_step`	Agent completes a reasoning step	Records a thinking step
`on_llm_call`	An LLM API call is made	Records an API call record
`on_tool_call`	A tool is executed	Records a tool execution step
`on_agent_complete`	Agent finishes the task	Completes the trace

Graceful Degradation

All hook methods follow the fail-safe pattern:

Check if the thinking service is available
If unavailable, log a debug message and return None
If available, delegate to the service
Catch and log any exceptions without re-raising

This ensures that Context Graph issues never impact the critical agent execution path.

Tenant Configuration

Hooks check the tenant configuration before capturing:

config = get_tenant_context_graph_config(tenant_id)
if not config or not config.thinking_enabled:
    return None

Singleton Access

from context_graph.integration.orchestrator_hooks import (
    get_orchestrator_hooks,
    create_orchestrator_hooks,
)
 
# Create with custom service
hooks = create_orchestrator_hooks(thinking_service=service)
 
# Get singleton
hooks = get_orchestrator_hooks()

Kafka Producers & Consumers Metrics Bridge