agora-core

Governance primitives for multi-agent systems.

When your AI agents make decisions together — who votes? Who can veto? How do you prevent cascading failures? Most frameworks assume a single "boss agent" orchestrates everything. Agora takes a different approach: democratic governance.

"In 48-agent production runs, we observed 19-day periods where every agent reported healthy telemetry — but the goal had quietly drifted." — AutoGen #7487

Why governance?

Multi-agent systems without governance suffer from:

• Goal drift: Agents individually succeed but collectively diverge from the original intent
• Cascading failures: One agent's error propagates unchecked through the chain
• No accountability: When things go wrong, there's no decision trail to audit
• Budget explosion: Agent A spawns B×5, each spawns C×5 — costs spiral in seconds

Policy engines like AGT solve the top-down compliance problem (Cedar policies, identity verification). Agora solves the collective decision-making problem.

What's included

Agora vs AGT vs built-in guardrails

Install

pip install agora-core

# Or from source
git clone https://github.com/lawcontinue/agora-core.git
cd agora-core && pip install -e .

Quick Start

from agora.governance.governance_layer import GovernanceLayer
from agora.agents.crits.local_crit import LocalCrit, ReviewDecision

# Define a crit (reviewer agent)
class SafetyCrit(LocalCrit):
    def _analyze(self, task):
        if "delete" in task["action"].lower():
            return {"decision": ReviewDecision.REJECT,
                    "reasoning": "P0: destructive operation blocked",
                    "confidence": 1.0}
        return {"decision": ReviewDecision.APPROVE,
                "reasoning": "Operation safe",
                "confidence": 0.9}

# Set up governance
gov = GovernanceLayer()
gov.register_local_crit(SafetyCrit())

# Review a decision
result = gov.review_decision({
    "task_id": "TASK_001",
    "description": "Drop user table",
    "agent": "AdminAgent",
    "action": "delete users"
})

print(result.approved)   # False — veto triggered
print(result.reasoning)  # "P0: destructive operation blocked"

Run the full demo:

python examples/quickstart.py

Architecture

                    ┌──────────────┐
                    │  Governance  │
                    │    Layer     │
                    └──────┬───────┘
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │  Local   │ │  Local   │ │  Local   │
        │  Crit A  │ │  Crit B  │ │  Crit C  │
        └──────────┘ └──────────┘ └──────────┘
              │            │            │
              └─────┬──────┘────────────┘
                    ▼
            ┌──────────────┐
            │   Consensus  │ ← 2/3 majority
            │   reached?   │
            └──┬───────┬───┘
               │       │
            Yes ▼    No ▼
          Approve  ┌──────────────┐
                   │ Global Crits │ ← 3-crit panel
                   │   Voting     │    tie-break
                   └──────┬───────┘
                          ▼
                   ┌──────────────┐
                   │ HITL         │ ← Human escalation
                   │ (1-1-1 split)│    if needed
                   └──────────────┘

API Overview

GovernanceLayer

Central orchestrator. Register crits, review decisions, query precedents.

gov = GovernanceLayer()
gov.register_local_crit(MyCrit())

result = gov.review_decision(decision_dict)
result.approved       # bool
result.reasoning      # str
result.local_reviews  # dict of crit responses
result.global_votes   # dict if escalated

LocalCrit

Base class for domain-expert reviewers with veto power.

class MyCrit(LocalCrit):
    def _analyze(self, task) -> dict:
        return {
            "decision": ReviewDecision.APPROVE | ReviewDecision.REJECT,
            "reasoning": "...",
            "confidence": 0.0-1.0
        }

PrecedentDB

Find similar past decisions using Jaccard similarity.

from agora.governance.precedent_db import PrecedentDB

db = PrecedentDB()
db.add_precedent("PRJ-001", tags=["safety", "delete"], outcome="rejected")
matches = db.find_similar(tags=["safety", "remove"])  # → [("PRJ-001", 0.5)]

OperationClassifier

Classify operations by risk level.

from agora.governance.operation_classifier import OperationClassifier

classifier = OperationClassifier()
level = classifier.classify("rm -rf /")  # → RiskLevel.P0 (default-deny)

Design Principles

1. Veto over consensus — One strong objection beats weak agreement
2. Precedent over rules — Learn from past decisions, don't just follow static rules
3. Default-deny for P0 — Destructive operations are blocked unless explicitly approved
4. Zero dependencies — Pure Python. No sklearn, no Cedar, no LLM required for core logic
5. Framework-agnostic — Works with AutoGen, CrewAI, LangGraph, or any agent system

Production Stats

Battle-tested with a 7-agent team across 190+ governance decisions. Zero safety incidents. Every decision logged, every veto recorded, every precedent retrievable.

License

Apache 2.0 — see LICENSE.

Credits

Built by lawcontinue with the T-Mind agent family. Production-validated alongside Hippo distributed inference.