Skip to content

Opt Shared

opt_shared — mvp.opt_shared

Cluster: Uncategorised | Type: component | MCP Tools: None

Overview

Shared schema and utility library for the G6 self-optimisation system. Defines the typed data structures used by opt_speed, opt_cost, opt_quality, and opt_meta: workflow DAGs, intervention actions, bottleneck analysis, resource snapshots, and intervention history. It also exposes OptSharedBlock operations for DAG validation, deterministic topological sorting, and safe conversion of optimisation interventions into pipeline config deltas.

When to use:

  • When building a new optimisation component that integrates with the opt_* stack
  • When serialising or deserialising workflow traces and intervention records
  • When validating traced workflow DAGs before analysis or reconfiguration
  • When converting approved optimisation interventions into pipeline config deltas
  • When extending the intervention action catalogue

Works well with: opt_speed, opt_cost, opt_quality, opt_meta, llm_router

Public API

OptSharedInput(BaseModel)

Input for shared optimisation operations.

Field Type Default
op str required
steps list[dict[str, Any]] Field(default_factory=list)
interventions list[dict[str, Any]] Field(default_factory=list)
parameters dict[str, Any] Field(default_factory=dict)

OptSharedOutput(BaseModel)

Output from shared optimisation operations.

Field Type Default
op str ''
result dict[str, Any] Field(default_factory=dict)
errors list[str] Field(default_factory=list)
message str ''
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card list[str] Field(default_factory=list)
evidence dict[str, Any] Field(default_factory=dict)
request_id str \| None None
task_id str \| None None
run_id str \| None None

OptSharedBlock(AIBlock)

Shared optimisation utilities: DAG validation, topological sort, actuation.

Methods:

infer(input: OptSharedInput) -> Result[OptSharedOutput]

VerificationResult(BaseModel)

Result of a verification step (e.g. CEGIS, type-check, unit-test).

Field Type Default
method str ''
passed bool True
details str ''
degraded bool False
degradation_reason str \| None None

ResourceSnapshot(BaseModel)

Point-in-time snapshot of resource bounds.

Field Type Default
max_execution_seconds float \| None None
max_tokens_per_minute int \| None None
max_tokens_per_hour int \| None None
max_cost_per_day float \| None None

WorkflowStep(BaseModel)

A single step inside a traced workflow execution.

Field Type Default
step_index int required
component_name str required
operation str required
kind StepKind 'deterministic'
latency_ms float 0.0
input_tokens int 0
output_tokens int 0
llm_model str ''
cost_usd float 0.0
success bool True
confidence float 1.0
verification VerificationResult Field(default_factory=VerificationResult)
error_message str ''
depends_on tuple[int, ...] ()

WorkflowTrace(BaseModel)

Complete trace of a workflow execution.

Field Type Default
trace_id str required
workflow_name str required
timestamp str required
steps tuple[WorkflowStep, ...] ()
total_latency_ms float 0.0
total_cost_usd float 0.0
total_input_tokens int 0
total_output_tokens int 0
quality_score float 1.0
resource_bounds ResourceSnapshot Field(default_factory=ResourceSnapshot)
metadata dict[str, str] Field(default_factory=dict)

Intervention(BaseModel)

A proposed optimisation intervention.

Field Type Default
intervention_id str required
intervention_type Literal['speed', 'cost', 'quality', 'all'] 'all'
action InterventionAction required
target_step int 0
target_component str ''
estimated_latency_delta_ms float 0.0
estimated_cost_delta_usd float 0.0
estimated_quality_delta float 0.0
confidence float 0.5
risk_level Literal['non_destructive', 'destructive', 'irreversible'] 'non_destructive'
requires_human_approval bool False
conflicts_with tuple[str, ...] ()
complements tuple[str, ...] ()
prerequisite_of tuple[str, ...] ()
rationale str ''
evidence str ''
evidence_basis EvidenceBasis 'heuristic_only'
uncertainty_reason str \| None 'recommendation is heuristic, not derived from a verified trace signal'
required_human_action str \| None 'review the recommendation against evidence before applying; this is a heuristic suggestion, not a verified result'

Bottleneck(BaseModel)

Identified bottleneck in a workflow trace.

Field Type Default
step_index int required
component_name str required
dimension Literal['latency', 'cost', 'quality', 'tokens'] 'latency'
severity Literal['low', 'medium', 'high', 'critical'] 'medium'
measured_value float 0.0
expected_value float 0.0
proportion float 0.0
description str ''

ImpactEstimate(BaseModel)

Estimated impact of a set of interventions.

Field Type Default
latency_reduction_pct float 0.0
cost_reduction_pct float 0.0
quality_improvement_pct float 0.0
confidence float 0.5

AutoApplyAction(BaseModel)

Record of an automatically applied intervention.

Field Type Default
intervention_id str required
applied_at str required
success bool True
error str ''
rollback_available bool False

OptimizationReport(BaseModel)

Full optimization report for a single trace.

Field Type Default
report_id str required
trace_id str required
analyzer str required
bottlenecks tuple[Bottleneck, ...] ()
interventions tuple[Intervention, ...] ()
estimated_impact ImpactEstimate Field(default_factory=ImpactEstimate)
auto_applied tuple[AutoApplyAction, ...] ()

InterventionOutcome(BaseModel)

Predicted vs actual outcome of an intervention.

Field Type Default
predicted_latency_delta_ms float 0.0
predicted_cost_delta_usd float 0.0
predicted_quality_delta float 0.0
actual_latency_delta_ms float 0.0
actual_cost_delta_usd float 0.0
actual_quality_delta float 0.0

Methods:

reward() -> float

Scalar reward combining latency, cost, and quality improvements.

InterventionRecord(BaseModel)

Mutable record of an applied intervention.

Field Type Default
record_id str required
intervention_id str required
action InterventionAction required
target_component str required
applied_at str required
trace_id_before str required
trace_id_after str ''
outcome InterventionOutcome \| None None
success bool \| None None
arm_key str ''

InterventionHistory(BaseModel)

Mutable history of intervention applications.

Field Type Default
records list[InterventionRecord] Field(default_factory=list)

Methods:

record_application(record: InterventionRecord) -> None

Append an intervention record.

record_outcome(record_id: str, outcome: InterventionOutcome, success: bool) -> None

Set outcome and success flag on a record identified by record_id.

best_arm() -> str | None

Returns the arm with the highest mean reward over all history.

success_rate(arm_key: str = '') -> float

Success rate of records with success is not None.

ReconfigurationAction(BaseModel)

An action to reconfigure a pipeline topology.

Field Type Default
action Literal['insert_before', 'insert_after', 'replace_params', 'remove_step', 'wrap_parallel'] required
target_step_index int required
new_step dict \| None None
param_updates dict[str, Any] Field(default_factory=dict)

Functions

validate_dag(steps: list[WorkflowStep]) -> list[str]

Return a list of error strings describing DAG violations.

topological_sort(steps: list[WorkflowStep]) -> list[WorkflowStep]

Return steps in topological order (dependencies before dependents).