Skip to content

Cortex API (semvec.cortex)

Multi-agent coordination primitives. Install with pip install "semvec[cortex]".

SemvecAgentNetwork

Container for multiple SemvecAgent objects aggregated into one SemvecCortexObserver.

from semvec.cortex import SemvecAgentNetwork, AttentionAggregation

net = SemvecAgentNetwork(
    aggregation_strategy=AttentionAggregation(),
    enable_feedback=True,
    feedback_strength=0.3,
    max_instances=10,
)
net.add_local_instance("analyst")
net.add_local_instance("planner")
net.process_input("analyst", "quarterly revenue is up 23%")
state = net.get_network_state()

Methods

Method Purpose
add_instance(id) / add_local_instance(id) Register a new agent.
remove_local_instance(id) Unregister.
process_input(instance_id, text) Drive one agent; MetaPSS updates internally per call.
get_network_state() -> dict Snapshot coherence + per-agent metrics.

Attributes

Attribute Type
local_instances dict[str, SemvecAgent]
meta_pss SemvecCortexObserver
enable_feedback bool
feedback_strength float
max_instances int

SemvecAgent

Per-agent PSS state.

from semvec.cortex import SemvecAgent

# `dimension=` propagates to the inner SemvecState. Use 768 for
# mpnet (recommended for German/multilingual), 384 for MiniLM.
# Default is 384.
inst = SemvecAgent("agent_a", embedder=my_embedder, dimension=768)
assert inst.pss_state.config.dimension == 768

result = inst.process_input("hello")
# or bypass text embedding:
result = inst.process_input_embedding(vector, "hello")

Missing embedder in process_input raises RuntimeError with guidance to either set embedder= or call process_input_embedding directly. Embedding-dimension mismatches in process_input_embedding raise ValueError (≥ 0.3.2 — pre-fix this was a Rust assertion panic that aborted the interpreter).

process_input(text, global_feedback=None) -> dict

Parameter Type Default Description
text str — Input text. Embedded with the agent's injected embedder.
global_feedback list[float] | np.ndarray | None None Optional meta-PSS / cortex feedback vector. When supplied, the input embedding is blended via (1 − 0.1) · embedding + 0.1 · feedback before the PSS update. Use this with SemvecCortexService.get_feedback_for_agent() to align an agent with the network consensus.

Returns the same dict as SemvecState.update() plus the per-agent keys instance_id, local_coherence, global_influence, state_vector, interaction_count.

process_input_embedding(embedding, text, global_feedback=None) -> dict

Same signature minus the embedder call. Use when you already have a normalised vector (e.g. batch-embedded offline).

Attributes

instance_id, local_coherence (read / write), interaction_count, global_influence (read / write), last_update_time, min_coherence_threshold, pss_state (read / write — used by SemvecCortexService when hydrating from a store).

SemvecCortexObserver

Aggregator for a collection of locals.

from semvec.cortex import SemvecCortexObserver, WeightedAverageAggregation

meta = SemvecCortexObserver(
    aggregation_strategy=WeightedAverageAggregation(dimension=384),
    dimension=384,
)
meta.update_global_state([local_a, local_b, local_c])
feedback = meta.get_feedback_for_instance("agent_a")
Attribute Type
global_state list[float]
global_coherence float
network_resonance float
beta_global float
update_count int

Aggregation strategies

WeightedAverageAggregation

aggregate(local_states, weights=None) — simple weighted mean with automatic None → uniform weights.

AttentionAggregation

Query/key/value attention with learnable projections. Constructed with AttentionAggregation(attention_dim=64, dimension=384).

Documented numeric drift

Projection matrices are seeded differently in pss (NumPy PCG64) and semvec (Rust StdRng), so byte-level aggregate outputs diverge even at the same seed. Structural invariants (shape, determinism per instance, softmax sum = 1) match exactly. See MIGRATION.md §Numerical Fidelity.

ConsensusEngine + ConsensusProposal + ConsensusLevel

from semvec.cortex import ConsensusEngine, ConsensusLevel
import numpy as np

engine = ConsensusEngine("me", "net_0")
for voter in ("cto", "sre", "fin_ops"):
    engine.register_instance(voter, weight=1.0)

proposal = engine.create_proposal(
    "state_update",
    proposed_state=np.zeros(8),
    rationale="move to Postgres 16",
    consensus_level=ConsensusLevel.QUALIFIED_MAJORITY,
    voting_timeout=300.0,
)
engine.vote_on_proposal(proposal.proposal_id, True, "cto")
engine.vote_on_proposal(proposal.proposal_id, True, "sre")
engine.vote_on_proposal(proposal.proposal_id, False, "fin_ops")
print(proposal.calculate_consensus())   # (True, 2/3)

ConsensusLevel is a string enum with members SIMPLE_MAJORITY, QUALIFIED_MAJORITY, UNANIMOUS, WEIGHTED_VOTE, ADAPTIVE_THRESHOLD. Threshold semantics match pss exactly.

Shared-state semantics

create_proposal returns a ConsensusProposal linked to the engine's internal storage. Voting through engine.vote_on_proposal(proposal_id, …) is reflected in the returned object. Once consensus is reached, the proposal is moved from active_proposals to proposal_history and further votes on the same ID return False.

Quorum semantics (≥ 0.3.2)

create_proposal snapshots the engine's known_instances voter pool at proposal-creation time. calculate_consensus then measures positive_weight / total_voter_pool_weight — not positive / votes_cast_so_far. Concretely: with 3 registered instances and SimpleMajority, the first YES gives ratio 1/3 (not reached), the second YES gives 2/3 > 0.5 (reached), and the third vote is silently rejected because the proposal was already finalised on vote #2.

Pre-0.3.2 the denominator was votes.len(), so the first local YES already produced 1/1 = 1.0 and tripped SimpleMajority before any remote instance could vote — remote votes were silently dropped. The pre-fix behaviour is still preserved when a ConsensusProposal is constructed directly without an engine (parity-test path).

StateVectorPacket + TransferType

Inter-agent state transfer with SHA-256/16 checksumming.

from semvec.cortex import StateVectorPacket, TransferType
import numpy as np

packet = StateVectorPacket(
    packet_id="pkt_1",
    source_instance="agent_a",
    target_instance="agent_b",
    transfer_type=TransferType.FULL_STATE,
    state_vector=np.arange(16, dtype=np.float64),
)
blob = packet.serialize()              # bytes (JSON)
restored = StateVectorPacket.deserialize(blob)
assert restored.verify_integrity()

TransferType members: FULL_STATE, PARTIAL_STATE, SEMANTIC_DELTA, MEMORY_FRAGMENT, COHERENCE_FIELD.

Bit-exact serialisation (≥ 0.3.2). The wire format carries a state_vector_bits: [u64, …] field with the IEEE 754 bit pattern of the state vector alongside the human-readable float array. Deserialisation prefers the bit pattern, so verify_integrity() returns True after a serialize → deserialize round-trip even on values that JSON's float encoder would otherwise round to a slightly different representation. Pre-0.3.2 the integrity check failed on freshly serialised packets because of that lossy round-trip.

SemvecCortexService

Cortex-as-a-service facade — a 1:1 port of pss.integrations.meta_pss_service.MetaPSSService. Coordinates multiple agents (1 per task/tenant) via attention-based or weighted-average aggregation, and exposes a global feedback vector that individual agents can mix into their updates.

from semvec.cortex import SemvecCortexService

svc = SemvecCortexService(
    pss_store=None,              # optional async store (see below)
    aggregation="attention",      # or "weighted"
    dimension=384,
)

Constructor

Argument Type Default Description
pss_store Protocol (async) or None None Any object exposing async list_active_states() -> Iterable[tuple[str, SemvecState]]. When None, the service runs fully in-memory and aggregates whatever was fed via register_agent() + process_input().
aggregation "attention" \| "weighted" "attention" Aggregation strategy. Any other value raises ValueError.
dimension int 384 Shared embedding dimension for agents and the aggregator.

Methods

  • register_agent(agent_id) -> bool — register a new agent on the in-memory network; returns False if already known.
  • process_input(agent_id, text) -> dict — drive one turn of the named agent (in-memory path; needs an embedder).
  • async update_global_state() -> dict — aggregate all active agents. Reads via pss_store.list_active_states() when a store is set, otherwise from the local cache populated by register_agent. Returns a dict with global_state, global_coherence, network_resonance, active_instances.
  • async get_global_context() -> str — human-readable summary for system prompts ("Global Coherence: 0.73, Network Resonance: 0.61, Active Agents: 3").
  • get_feedback_for_agent(agent_id) -> np.ndarray — per-agent feedback vector of shape (dimension,). Agents mix this into their updates via SemvecAgent.process_input(text, global_feedback=fb).
  • _calculate_coherence(state) -> float / _calculate_influence(state) -> float — expose the scoring math used for weighting aggregates (pure compute on a SemvecState).

Example with a fake store

class InMemoryStore:
    def __init__(self, states): self._states = states
    async def list_active_states(self): return self._states

svc = SemvecCortexService(pss_store=InMemoryStore([("alpha", s1), ("beta", s2)]))
result = await svc.update_global_state()
feedback = svc.get_feedback_for_agent("alpha")