Cortex API (semvec.cortex)

Multi-agent coordination primitives — run several agents that share an aggregated view, vote on proposals, and exchange checksummed state vectors. Install with pip install "semvec[cortex]".

SemvecAgentNetwork

Container for multiple SemvecAgent objects aggregated into one SemvecCortexObserver.

snippet — embed in your own setup, network agents need an embedder via process_input_embedding

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")
# Network-managed agents are created internally without an embedder;
# drive them with precomputed embeddings or use SemvecAgent directly.
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; the aggregated observer updates internally per call.
get_network_state() -> dict	Snapshot coherence + per-agent metrics.
get_feedback_for_agent(agent_id) -> np.ndarray	Per-agent feedback vector that the agent can blend into its next update.

Attributes

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

SemvecAgent

Per-agent state. The embedder must be an object that exposes get_embedding(text) -> np.ndarray and get_dimension() -> int.

import numpy as np
from sentence_transformers import SentenceTransformer
from semvec.cortex import SemvecAgent

model = SentenceTransformer("all-MiniLM-L6-v2")

class STEmbedder:
    def get_dimension(self): return 384
    def get_embedding(self, text):
        return np.asarray(model.encode(text), dtype=np.float64)

my_embedder = STEmbedder()

# Use 768 for mpnet (recommended for German/multilingual), 384 for MiniLM.
inst = SemvecAgent(instance_id="agent_a", embedder=my_embedder)
assert inst.pss_state.config.dimension == 384

result = inst.process_input("hello")
# or bypass text embedding:
vector = my_embedder.get_embedding("hello")
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.

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 cortex feedback vector. When supplied, the input embedding is blended via (1 − 0.1) · embedding + 0.1 · feedback before the update. Use this with SemvecAgentNetwork.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 (for example 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 agents.

snippet — assumes local_a/local_b/local_c are SemvecAgent instances from the SemvecAgent block above

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).

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.

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

create_proposal snapshots the engine's known_instances voter pool at proposal-creation time. calculate_consensus measures positive_weight / total_voter_pool_weight — not positive / votes_cast_so_far. Concretely: with 3 registered instances and SIMPLE_MAJORITY, 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.

StateVectorPacket + TransferType

Inter-agent state transfer with checksummed integrity.

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.

The wire format carries an IEEE 754 bit-pattern alongside the human-readable float array, 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.

SemvecCortexService

Cortex-as-a-service facade. 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).

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")

See also

• Quickstart — 5-minute REST + library walk-through
• Cortex (multi-agent) — user-guide page that contextualises this API
• Architecture — abstract component model