syrin 0.11.0

Python library for building AI agents with budget management, declarative definitions, and production-ready observability.

The Python Library for Production AI Agents

Budget control · Persistent memory · Multi-agent orchestration · Built-in observability

Website · Docs · Discord · Reddit · YouTube

---

The Problem with AI Agents in Production

Most AI agent frameworks hand you primitives and wish you luck. You prototype in a weekend, then spend weeks bolting on cost tracking, memory, observability, and guardrails before anything goes near production.

Syrin is built differently. Budget control, memory, observability, multi-agent orchestration, and safety guardrails are first-class features — not afterthoughts.

---

Quick Installation

pip install syrin

30-Second Quickstart

from syrin import Agent, Budget, Model
from syrin.enums import ExceedPolicy

class Analyst(Agent):
    model  = Model.OpenAI("gpt-4o-mini", api_key="...")
    budget = Budget(max_cost=0.10, exceed_policy=ExceedPolicy.STOP)
    system_prompt = "You are a precise financial analyst."

result = Analyst().run("Summarise Q3 revenue trends from the attached report")

print(result.content)
print(f"Cost: ${result.cost:.6f}  |  Tokens: {result.tokens}  |  Remaining: ${result.budget_remaining:.4f}")

That's it. The agent hard-stops at $0.10. No surprise invoices, no extra code.

---

What You Can Build

Use Case	Key Capabilities
Research pipelines	Parallel agents gather sources; consensus topology validates findings
Customer support bots	Session memory, PII guardrails, handoff to human when confidence is low
Document intelligence	RAG over PDFs/repos, structured output, budget per document
Financial analysis	Hard cost caps, audit logs, type-safe structured results
Voice AI assistants	ElevenLabs/Deepgram integration, memory across calls
Compliance review	PII scanning, output validation, full provenance records
Code review agents	Knowledge base over your codebase, tool use, structured findings
Autonomous schedulers	Cron/webhook/queue triggers, checkpoints for crash recovery

---

What Makes Syrin Different

Budget Control — The Industry Gap

Every other library treats cost as a monitoring concern. Syrin treats it as a runtime constraint. Agents check their budget before every LLM call and enforce limits with the policy you choose.

from syrin import Agent, Budget, Model, RateLimit
from syrin.enums import ExceedPolicy
from syrin.budget import BudgetThreshold

class ProductionAgent(Agent):
    model  = Model.OpenAI("gpt-4o", api_key="...")
    budget = Budget(
        max_cost=1.00,                        # Hard cap per run
        reserve=0.10,                         # Hold back for final reply
        exceed_policy=ExceedPolicy.STOP,      # STOP | WARN | IGNORE | SWITCH
        rate_limits=RateLimit(
            hour=10.00,                       # $10/hour across all runs
            day=100.00,                       # $100/day
            month=2000.00,                    # $2,000/month
        ),
        thresholds=[
            BudgetThreshold(at=80, action=lambda ctx: alert_ops_team(ctx)),
        ],
    )

Pre-call estimation, post-call actuals, threshold callbacks, rate-window enforcement — all declarative, zero boilerplate.

---

Multi-Agent Orchestration — 5 Topologies

from syrin import Agent, Budget, Model
from syrin.swarm import Swarm, SwarmConfig, BudgetPool
from syrin.enums import SwarmTopology

pool = BudgetPool(total=5.00)  # $5 shared; no agent can exceed its slice

class Researcher(Agent):
    model = Model.OpenAI("gpt-4o", api_key="...")

class FactChecker(Agent):
    model = Model.Anthropic("claude-sonnet-4-5", api_key="...")

class Writer(Agent):
    model = Model.OpenAI("gpt-4o-mini", api_key="...")

swarm = Swarm(
    agents=[Researcher, FactChecker, Writer],
    config=SwarmConfig(
        topology=SwarmTopology.ORCHESTRATOR,  # LLM routes work dynamically
        budget_pool=pool,
    ),
)

result = swarm.run("Research and write a report on battery technology trends")
print(result.content)
print(result.cost_breakdown)   # Per-agent cost breakdown
print(result.budget_report)    # Pool utilisation

Five topologies — one Swarm class:

Topology	What it does
ORCHESTRATOR	First agent routes tasks to the rest dynamically
PARALLEL	All agents run concurrently; results merged
CONSENSUS	Multiple agents vote; winner selected by strategy
REFLECTION	Producer–critic loop until quality threshold is met
WORKFLOW	Sequential, parallel, branch, and dynamic fan-out steps

---

Persistent Memory — 4 Types, Extensible Backends

from syrin import Agent, Model
from syrin.enums import MemoryType

agent = Agent(model=Model.OpenAI("gpt-4o-mini", api_key="..."))

# Store — persisted across sessions
agent.remember("User is a TypeScript engineer at a fintech startup", memory_type=MemoryType.FACTS)
agent.remember("Previous session: reviewed authentication architecture", memory_type=MemoryType.HISTORY)

# Recall — semantic search over stored memories
memories = agent.recall("user background", limit=5)

# Forget — when facts are outdated
agent.forget("previous role title")

Memory Type	Stores
FACTS	Identity and preferences — user name, role, language, persistent facts
HISTORY	Past events and conversations — chronological, session context
KNOWLEDGE	General knowledge and concepts — ideal for vector/semantic search
INSTRUCTIONS	Skills and how-to knowledge — workflows, user preferences on process

Backends — swap with one line:

from syrin import Agent, Model
from syrin.memory import Memory, QdrantConfig, RedisConfig, PostgresConfig
from syrin.enums import MemoryBackend

# SQLite — zero config, persists to disk (default for most projects)
agent = Agent(model=model, memory=Memory(backend=MemoryBackend.SQLITE, path="~/.syrin/memory.db"))

# Qdrant — vector search, ideal for semantic recall at scale
agent = Agent(model=model, memory=Memory(
    backend=MemoryBackend.QDRANT,
    qdrant=QdrantConfig(url="https://your-cluster.qdrant.io", api_key="..."),
))

# Redis — fast cache with optional TTL, good for short-lived session memory
agent = Agent(model=model, memory=Memory(
    backend=MemoryBackend.REDIS,
    redis=RedisConfig(host="localhost", port=6379, ttl=3600),
))

# PostgreSQL — production-grade, pgvector for embeddings
agent = Agent(model=model, memory=Memory(
    backend=MemoryBackend.POSTGRES,
    postgres=PostgresConfig(host="localhost", database="agents", user="...", password="..."),
))

Backend	Best for
MEMORY	Testing and ephemeral runs — no setup, no persistence
SQLITE	Local development and single-process production — zero config
QDRANT	Semantic search at scale — cloud-ready vector database
CHROMA	Lightweight local vector search — fast to set up
REDIS	Session-scoped memory with TTL — low latency
POSTGRES	Full production — ACID, pgvector, multi-agent shared store

---

Observability — 72+ Lifecycle Hooks

Every LLM call, tool invocation, budget event, memory operation, and handoff emits a typed hook. Attach handlers at the agent level — no patching, no monkey-patching.

from syrin.enums import Hook

agent.events.on(Hook.AGENT_RUN_END,      lambda ctx: metrics.record(ctx.cost, ctx.tokens))
agent.events.on(Hook.BUDGET_THRESHOLD,   lambda ctx: pagerduty.alert(f"Budget at {ctx.percentage}%"))
agent.events.on(Hook.TOOL_CALL_END,      lambda ctx: logger.info(f"Tool {ctx.name} → {ctx.duration_ms}ms"))
agent.events.on(Hook.MEMORY_RECALL,      lambda ctx: trace.span("recall", memories=ctx.count))

Or enable full tracing from the CLI — no code changes:

python my_agent.py --trace

---

Live Terminal Debugger — Pry

A Rich-based TUI that lets you step through agent execution, inspect state, and set breakpoints — directly in your terminal.

from syrin.debug import Pry

pry = Pry()
pry.attach(agent)
agent.run("Analyse this dataset")
# [e] events  [t] tools  [m] memory  [g] guardrails  [p] pause  [n] step  [q] quit

---

Guardrails & Safety

from syrin import Agent, Model
from syrin.guardrails import PIIGuardrail, LengthGuardrail, ToolOutputValidator
from pydantic import BaseModel

class SafeAgent(Agent):
    model      = Model.OpenAI("gpt-4o-mini", api_key="...")
    guardrails = [
        PIIGuardrail(redact=True),          # Redact emails, phones, SSNs, card numbers
        LengthGuardrail(max_length=4000),   # Enforce output length
    ]

result = SafeAgent().run("Process: please call me at 555-123-4567")
# result.content → "please call me at ***-***-****"
# result.report.guardrail.passed → False

---

Production-Ready from Day One

One-line HTTP serving:

agent.serve(port=8000, enable_playground=True)
# → POST /chat   POST /stream   GET /playground

Crash-proof checkpoints:

from syrin.checkpoint import CheckpointConfig

agent = Agent(
    model=model,
    checkpoint_config=CheckpointConfig(dir="/tmp/checkpoints", auto_save=True),
)
result = agent.run("Begin long analysis...")
# Crash? Resume exactly where it left off:
agent.load_checkpoint("analysis-run-1")

Event-driven triggers:

from syrin.watch import CronProtocol, WebhookProtocol

agent.watch(CronProtocol(cron="0 9 * * *"), task="Send morning briefing")
agent.watch(WebhookProtocol(path="/events"), task="Process incoming event")

---

Real-World Examples

Research Pipeline — Parallel Agents with Consensus

from syrin import Agent, Budget, Model
from syrin.swarm import Swarm, SwarmConfig, ConsensusConfig
from syrin.enums import SwarmTopology, ConsensusStrategy

class ResearchAgent(Agent):
    model = Model.OpenAI("gpt-4o", api_key="...")
    system_prompt = "Research the given topic with citations."

swarm = Swarm(
    agents=[ResearchAgent, ResearchAgent, ResearchAgent],  # 3 independent researchers
    config=SwarmConfig(
        topology=SwarmTopology.CONSENSUS,
        consensus=ConsensusConfig(
            min_agreement=0.67,
            voting_strategy=ConsensusStrategy.MAJORITY,
        ),
        budget_pool=BudgetPool(total=3.00),
    ),
)

result = swarm.run("What are the main risks of lithium-ion batteries at scale?")
print(result.content)           # Consensus answer
print(result.cost_breakdown)    # Cost per agent

---

Customer Support — Memory + Handoffs + Guardrails

from syrin import Agent, Budget, Model
from syrin.guardrails import PIIGuardrail
from syrin.enums import MemoryType

class SupportAgent(Agent):
    model      = Model.OpenAI("gpt-4o-mini", api_key="...")
    budget     = Budget(max_cost=0.05)    # $0.05 per conversation turn
    guardrails = [PIIGuardrail(redact=True)]
    system_prompt = "You are a helpful customer support agent."

class EscalationAgent(Agent):
    model = Model.OpenAI("gpt-4o", api_key="...")
    system_prompt = "You handle escalated support cases requiring senior judgment."

agent = SupportAgent()
agent.remember(f"Customer {user_id}: premium plan, joined 2023", memory_type=MemoryType.FACTS)

result = agent.run(user_message)

if result.confidence < 0.6:
    result = agent.handoff(EscalationAgent, context=result.content)

---

Document Intelligence — RAG + Structured Output

from syrin import Agent, Model
from syrin.knowledge import Knowledge
from syrin.enums import KnowledgeBackend
from pydantic import BaseModel

class ContractRisk(BaseModel):
    risk_level: str           # low | medium | high | critical
    key_clauses: list[str]
    recommended_action: str

kb = Knowledge(
    sources=["path/to/contracts/"],
    backend=KnowledgeBackend.QDRANT,
    embedding_provider="openai",
)

class ContractReviewer(Agent):
    model       = Model.OpenAI("gpt-4o", api_key="...")
    budget      = Budget(max_cost=0.25)
    knowledge   = kb
    output_type = ContractRisk

result = ContractReviewer().run("Review the indemnification clause in contract-2024-07.pdf")
risk: ContractRisk = result.output   # Guaranteed typed output
print(f"Risk: {risk.risk_level} — {risk.recommended_action}")

---

Installation

# With OpenAI support
pip install syrin[openai]

# With Anthropic support
pip install syrin[anthropic]

# Multi-modal — voice, documents, vector stores
pip install syrin[voice,pdf,vector]

# Full install
pip install syrin[openai,anthropic,serve,vector,postgres,pdf,voice]

---

Why Syrin

Capability	Syrin	DIY / Other Libraries
Budget enforcement	Declarative, pre-call + post-call	Not available or manual
Rate windows	hour / day / month built-in	Build and persist yourself
Threshold callbacks	BudgetThreshold(at=80, ...)	Write from scratch
Shared budget pools	Thread-safe BudgetPool	Implement locking
Memory (4 types)	Built-in, auto-managed, backend-agnostic	Manual setup
Multi-agent (5 topologies)	Single Swarm class	Complex orchestration code
Lifecycle hooks	72+ typed events	Logging + parsing
Live debugger	Rich TUI (Pry)	Parse log files
Guardrails	PII, length, content, output validation	Per-project code
Checkpoints	Auto-save, crash recovery	DIY
RAG / Knowledge	GitHub, docs, PDFs, websites	Manual indexing pipeline
Structured output	Guaranteed Pydantic / JSON	Parse + validate manually
Serving	One-line HTTP + playground	Build Flask/FastAPI wrapper
Type safety	StrEnum everywhere, mypy strict	String literals

---

Documentation

Guide	Description
Quick Start	Your first agent in 5 minutes
Budget Control	Caps, rate limits, thresholds, shared pools
Memory	4 types, backends, decay curves
Multi-Agent Swarms	5 topologies, A2A messaging, budget delegation
Observability & Hooks	72+ events, tracing, Pry debugger
Guardrails	PII, length, content filtering, output validation
Knowledge / RAG	Ingestion, chunking, retrieval
Serving	HTTP API, streaming, playground
Checkpoints	State persistence, crash recovery
Examples	Runnable code for every use case

---

Join the Community

We are building a community of engineers shipping AI agents to production. Ask questions, share what you have built, and help shape the roadmap.

     

Channel	What's there
Discord	Real-time help, showcase your agents, roadmap discussion
Reddit — r/syrin_ai	Longer posts, tutorials, use-case deep dives
YouTube — @syrin_dev	Walkthroughs, feature demos, production patterns
GitHub Discussions	RFCs, architecture questions, feature requests
Website	Product overview, roadmap, changelog

---

Contributing

Contributions are welcome. See CONTRIBUTING.md for guidelines on setting up the dev environment, running tests, and submitting pull requests.

---

License

MIT — see LICENSE for details.

---

Built for engineers who ship AI to production.