API reference¶
Everything below is importable from poincare_orchestra unless noted. Public API changes are additive and listed in the changelog.
Scenario model¶
Scenario model: a use case is data, not code.
A scenario bundles the domain hierarchy and dated facts that seed the memory,
the specialist roles and their capability profiles, example tasks with the
facets a good answer must cover, and the controller policy. Everything the
orchestrator does is driven by it, so a new use case is a new scenario file
(JSON or YAML; see :meth:Scenario.from_file).
module-attribute
¶
The five synthesis objectives a policy may weight or constrain.
module-attribute
¶
Accepted kinds in a ControllerPolicy.decay spec.
Bases: BaseModel
A complete use case: memory contents, roles, capability profiles, tasks and policy.
Validation runs on construction and on :meth:from_file: duplicate ids,
parents declared after their children, roles that reference unknown
profiles and tasks without facets are reported with their position.
class-attribute
instance-attribute
¶
policy: ControllerPolicy = Field(default_factory=ControllerPolicy)
The poincare_memory decay policy described by policy.decay.
build_memory(now: datetime | None = None, clock: Callable[[], datetime] | None = None, **kwargs: Any) -> Memory
Seed a fresh :class:poincare_memory.Memory with the knowledge items.
now anchors the relative dates (default: the current UTC time);
clock overrides the memory's clock; other keyword arguments go to
Memory (for example embedder=).
JSON-compatible dict with defaults and None fields omitted (the on-disk form).
Write the scenario as JSON or YAML; the format follows the suffix unless given.
YAML needs PyYAML (pip install 'poincare-orchestra[yaml]').
classmethod
¶
from_file(path: str | Path, format: FileFormat | None = None) -> Scenario
Load and validate a scenario from a JSON or YAML file.
Raises:
| Type | Description |
|---|---|
FileNotFoundError
|
if |
ValueError
|
if the suffix is not |
ValidationError
|
if the contents are inconsistent (see the class docstring). |
ImportError
|
for YAML without PyYAML installed. |
Bases: BaseModel
One record of the scenario memory.
parent builds the hierarchy (parents must be listed before their
children); days_ago dates a record relative to the run; the two
valid_* fields give it a validity window for the validity decay kind.
class-attribute
instance-attribute
¶
Age of a dated record; undated items are structural knowledge and count as current.
Bases: BaseModel
An example question with the facets (substrings) a good answer must contain.
class-attribute
instance-attribute
¶
expected_facets: list[str] = Field(default_factory=list, description='substrings a good answer must contain')
Bases: BaseModel
Budget, retrieval, objective weights, epsilon constraints and temporal policy of a scenario.
class-attribute
instance-attribute
¶
objectives: dict[str, float] = Field(default_factory=lambda: {'coverage': 0.4, 'density': 0.2, 'agreement': 0.1, 'recency': 0.15, 'cost': 0.15})
Weights w_i of eq. 7; keys must be objective names.
class-attribute
instance-attribute
¶
Epsilon thresholds ε_j of eq. 7 in [0, 1]; keys must be objective names.
class-attribute
instance-attribute
¶
none | half_life:<hours> | ebbinghaus:<hours> | validity | half_life:<h>+validity
Build a poincare_memory decay policy from a spec string.
Grammar: none | half_life:<hours> | ebbinghaus:<hours> | validity,
and any of these joined with + for a product (for example
half_life:2160+validity). Hours default to 720 (half-life) and 168
(Ebbinghaus stability) when omitted.
Raises:
| Type | Description |
|---|---|
ValueError
|
if a kind is unknown or an hour value is not a positive number. |
load_scenario(name: str | Path) -> Scenario
A bundled scenario by name, or any scenario file by path (.json, .yaml, .yml).
Raises:
| Type | Description |
|---|---|
KeyError
|
if |
FileNotFoundError
|
if |
Roles and capability profiles¶
Capability profiles and agent roles (paper §IV-B capability matching, §IV-D capability profiling).
Schemas¶
Standardised schemas shared by every component (paper §IV-D "response standardization").
Bases: BaseModel
One unit of the decomposition (eq. 5) with its role, dependencies, gain, cost and budget.
class-attribute
instance-attribute
¶
Text used for retrieval; the question itself may carry a display-only focus label.
Bases: BaseModel
Output of the controller: query analysis plus the ordered sub-tasks.
Bases: BaseModel
Standardised provider output (paper §IV-D): content, claims, confidence, cost signals, refusal flag.
Bases: BaseModel
A response that broke its provider's baseline, or a provider exception (reason error: <type>).
Bases: BaseModel
Fused answer with the five objective scores, eq. 7 weighted sum, constraint status and uncertainty.
Bases: BaseModel
Everything one :meth:~poincare_orchestra.Orchestrator.run produced.
class-attribute
instance-attribute
¶
throttle_events: list[ThrottleEvent] = Field(default_factory=list)
The whole result (plan, responses, synthesis, allocations, events, trace) as JSON text.
Human-readable trace: question, plan, responses, throttle events, scores and the answer.
Controller¶
Metacognitive controller (paper §III-B, §III-C, §IV-B).
Four executive functions, each a small, inspectable step:
1. query analysis and intent recognition, with the memory hierarchy as the domain model;
2. hierarchical decomposition into coordinated sub-tasks Q -> {q_1..q_n} (eq. 5);
3. capability matching of sub-tasks to roles and profiles;
4. resource allocation by expected information gain per cost,
I*(q_i) = argmax_I G(q_i, I) / C(I) (eq. 6), run as an investment cycle:
invest, measure yield, re-invest the remainder.
The implementation is rule-based and offline so the control loop is transparent;
an LLM-backed planner can replace decompose without touching the rest.
module-attribute
¶
InvestmentScheduler(policy: ControllerPolicy, embedder: HashingEmbedder | None = None)
Allocate compute by expected information gain per cost (eq. 6), then re-invest after measuring yield.
expected_gain(plan: Plan, memory: Memory) -> dict[str, float]
Gain = relevance of what the memory holds for the sub-task × novelty versus the other sub-tasks.
allocate(plan: Plan, memory: Memory) -> dict[str, float]
Keep sub-tasks in order of gain/cost while they fit the budget, then split the budget by gain/cost.
Sub-tasks that do not fit are marked skipped with budget 0: the
allocation decision of eq. 6 made explicit instead of overspending.
staticmethod
¶
measure_yield(response: AgentResponse) -> float
Information yield proxy: distinct claims per 100 tokens (the paper's 'information density').
reinvest(plan: Plan, responses: list[AgentResponse], spent: float) -> dict[str, float]
Extra budget for the sub-tasks with the lowest yield (refinement targets).
dataclass
¶
QueryAnalysis(query: str, intent: str, complexity: float, domains: list[str], branch_ids: list[str], entity_ids: list[str], retrieved_ids: list[str], clauses: list[str])
Result of :func:analyze: intent, complexity, the memory branches and entities a query touches.
analyze(query: str, memory: Memory, k: int = 5) -> QueryAnalysis
Intent, complexity, domain branches and named entities for a query.
decompose(analysis: QueryAnalysis, memory: Memory, roles: list[AgentRole], profiles: list[CapabilityProfile], policy: ControllerPolicy) -> Plan
Split a query into facet sub-tasks (one per domain branch) and sequential clause sub-tasks, then match roles.
Records whose leading phrase (the text before ':' or '(') names an entity mentioned in the query.
Convention: entity records read "Athlete A: ...", "Ticket 4411: ...", "Release 2.4 (...)".
The ancestor one level below the root (the domain branch), or the node itself.
Ensemble¶
Provider-agnostic ensemble (paper §IV-D): providers, throttling detection, load balancing, failover.
Bases: Protocol
Anything with a name and a complete(request) -> AgentResponse method.
MockProvider(name: str, degrade_after: int | None = None, latency_s: float = 0.01, seed: int = 0, fail_after: int | None = None)
Deterministic provider that answers from the retrieved context.
It lets the whole orchestration run offline and in tests. degrade_after
simulates a throttled commercial model: after that many calls, answers
become truncated, slow and sometimes refused, which the detector should
catch and the router should route around. fail_after simulates an
outage: after that many calls every call raises RuntimeError.
LiteLLMProvider(name: str, model: str, temperature: float = 0.2, max_tokens: int = 600, timeout_s: float | None = None, **kwargs: Any)
Any model LiteLLM can reach (pip install 'poincare-orchestra[litellm]').
model is a LiteLLM model string such as openai/gpt-4o-mini,
anthropic/claude-haiku-5-5 or ollama/llama3; credentials come
from the environment variables LiteLLM reads for that provider. The prompt
asks for a JSON object {"answer": str, "claims": [{"text", "confidence",
"sources"}]}; plain text is accepted as a fallback. timeout_s is passed
to litellm.completion(timeout=...); a timeout raises and the router
treats it like any other provider error (refused response, throttle event,
failover). Extra keyword arguments go to litellm.completion unchanged.
build_messages(request: CompletionRequest) -> list[dict[str, str]]
The system and user messages sent for a request.
ThrottleDetector(window: int = 6, min_history: int = 2, latency_factor: float = 3.0, length_ratio: float = 0.4, density_drop: float = 0.35)
Flags responses whose latency, length, density or refusal pattern breaks the provider's baseline.
The baseline is the provider's own recent responses (the last window
that were not flagged); while a provider has fewer than min_history of
them, the pooled responses of all providers stand in for it. A response is
flagged when it is refused, when its latency exceeds latency_factor
times the baseline median, when its token count is below length_ratio
times the baseline median, or when its information density is more than
density_drop below the baseline median. Flagged responses do not enter
the baseline.
instance-attribute
¶
observe(response: AgentResponse) -> ThrottleEvent | None
Record a response; return a :class:ThrottleEvent if it breaks the baseline, else None.
Router(providers: list[Provider], profiles: list[CapabilityProfile], detector: ThrottleDetector | None = None, cooldown_calls: int = 3)
Chooses a provider for each role by capability profile and health; fails over on throttling.
Candidates are ordered by: the profile's preferred provider first, then
providers that are not cooling down, then the least loaded. A flagged
response or a provider exception puts the provider in cooldown: it is
skipped for the next cooldown_calls calls and the next candidate is
tried; the last response is returned if every candidate fails. When every provider is
cooling down they are tried anyway, least-degraded first, so a call never
fails only because of cooldowns. call never raises on a provider
exception: it is converted into a refused :class:AgentResponse and a
:class:ThrottleEvent with reason error: <exception type>.
Advance every cooldown by one call (called automatically at the end of each :meth:call).
call(request: CompletionRequest, profile_name: str) -> AgentResponse
Complete request with the best healthy provider for profile_name (see the class docstring).
parse_completion(text: str) -> tuple[str, list[Claim]]
Split a model reply into (answer, claims).
The reply is expected to contain a JSON object {"answer": str, "claims": [...]}
(code fences and surrounding prose are tolerated); otherwise the whole text is the
answer and there are no claims. Malformed claim entries are skipped.
Short replies of the form "I cannot/can't/am unable to help|assist|comply".
Unique-token ratio: a cheap proxy for the paper's bits-per-token notion.
Synthesis¶
Response synthesis (paper §III-D, §IV-E): fusion, multi-objective scoring with epsilon constraints, uncertainty from agreement, and iterative refinement.
dataclass
¶
synthesize(plan: Plan, responses: list[AgentResponse], recency_weights: dict[str, float], weights: dict[str, float], constraints: dict[str, float], spent: float, budget: float, refinements: int = 0) -> SynthesisResult
fuse_claims(responses: list[AgentResponse], merge_threshold: float = 0.85) -> list[Claim]
Merge near-duplicate claims across sources; confidence is the max, sources are pooled.
score_objectives(plan: Plan, responses: list[AgentResponse], claims: list[Claim], recency_weights: dict[str, float], spent: float, budget: float) -> Objectives
weighted_sum(obj: dict[str, float], weights: dict[str, float], constraints: dict[str, float]) -> tuple[float, bool, list[str]]
Paper eq. 7: maximise Σ w_i f_i subject to f_j ≥ ε_j.
agreement(responses: list[AgentResponse]) -> float
Mean pairwise overlap of claim sources across distinct providers (1 = every provider agrees).
compose_answer(plan: Plan, claims: list[Claim], responses: list[AgentResponse]) -> str
Orchestrator¶
The end-to-end loop: analyse → decompose → allocate → execute → synthesise → refine.
Bases: BaseModel
Facet recall and synthesis scores over a scenario's bundled tasks (see :meth:Orchestrator.evaluate).
class-attribute
instance-attribute
¶
Orchestrator(scenario: Scenario, providers: list[Provider] | None = None, policy: ControllerPolicy | None = None, memory: Memory | None = None, clock: Callable[[], datetime] | None = None, detector: ThrottleDetector | None = None)
Runs questions through a scenario: the eight steps of the paper's control loop.
With providers=None one offline :class:MockProvider is created per
provider name used by the scenario's capability profiles. Pass
:class:LiteLLMProvider instances (or any object satisfying
:class:Provider) for real models; their names should match the
provider field of the profiles so that capability matching routes to them.
run(question: str) -> RunResult
Answer one question; never raises on provider failures (they become refused responses and events).
evaluate() -> EvalReport
Run every bundled task and measure facet recall (expected substrings found in the answer and responses).
Command line¶
poincare-orchestra list | run | eval | inspect | export | validate.
Every command that takes a scenario accepts a bundled name (list prints
them) or the path to a .json / .yaml scenario file. Exit codes: 0 on
success, 1 when validate finds a problem, 2 when a scenario cannot be loaded.
Bundled scenarios¶
poincare_orchestra.scenarios.sprint_coaching, poincare_orchestra.scenarios.clinical_anatomy and poincare_orchestra.scenarios.support_kb each expose a SCENARIO constant (a Scenario) built from module-level KNOWLEDGE, PROFILES, ROLES and TASKS lists. load_scenario(name) returns that constant; the same scenarios are exported as YAML under scenarios/ in the repository.