enterprise-agentic-ai-framework

An enterprise governance framework for building single- and multi-agent AI systems in Python: authorization, guardrails, observability, secrets management, LLM gateway access, and a full production evaluation suite, all as one consistent stack instead of one-off code per project.

pip install enterprise-agentic-ai-framework

The import name is agentic_ai (the PyPI distribution name is longer for naming reasons, the package you actually import is not):

from agentic_ai.gateway import LiteLLMGateway

Status

This is an early release. The LLM gateway, the full evaluation suite, Memory & State, and Context Engineering are implemented today - everything else below is scaffolded (the module exists, it's empty) and not yet usable. This table will be kept current as modules land, not written once and left stale.

Module	Status
gateway - LLM gateway (LiteLLM proxy client)	✅ Implemented
evaluation - Agent/LLM/Tools/Multi-Agent/RAG/Security/Platform/Memory/Drift evaluation (48 metrics, see below)	✅ Implemented
memory - session/agent/short-term/working/long-term/semantic/episodic/procedural/document/shared memory & state (see below)	✅ Implemented
context - context engineering: assembly, write, select, compress, isolate (see below)	✅ Implemented
identity - authentication	⏳ Planned
governance - authorization (PEP/PDP)	⏳ Planned
guardrails - PII/secrets/injection/jailbreak detection	⏳ Planned
secrets - secrets management	⏳ Planned
observability - distributed tracing, structured audit	⏳ Planned
finops - LLM cost tracking	⏳ Planned
security - rate limiting, abuse detection (live enforcement)	⏳ Planned
compliance, audit, data_governance	⏳ Planned
monitoring, resilience, responsible_ai	⏳ Planned
core - agent/tool base classes, orchestrator	⏳ Planned

A naming note, not a contradiction: evaluation.memory and the top-level memory module are different things. evaluation.memory measures something (was a memory retrieval accurate/consistent?) from data you already collected. The top-level memory module described below is the live storage layer - the thing evaluation.memory would be measuring. Same relationship for evaluation.security (still planned as a top-level module) vs. the eventual live security enforcement layer.

Prerequisites

This library is a client, not a server. Before any of the examples below will work, you need a LiteLLM proxy already running somewhere reachable - agentic_ai.gateway never installs, starts, stops, or otherwise manages that process for you. Set it up once:

1. Install LiteLLM's proxy (a separate package from this library):

pip install 'litellm[proxy]'

2. Register at least one model. Create litellm_config.yaml - this example routes the model name gpt-4o-mini to OpenAI, reading the real provider key from an environment variable (never hardcode it in the YAML):

model_list:
  - model_name: gpt-4o-mini
    litellm_params:
      model: openai/gpt-4o-mini
      api_key: os.environ/OPENAI_API_KEY

Any provider LiteLLM supports works the same way - Anthropic, Azure OpenAI, Bedrock, a local Ollama model, etc.; only litellm_params changes. See LiteLLM's own docs for the full provider list.

3. Set the real provider key and start the proxy:

export OPENAI_API_KEY=sk-...
litellm --config litellm_config.yaml --port 4000

4. Confirm it's actually up before writing any Python against it:

curl http://localhost:4000/health/liveliness
# -> "I'm alive!"

If that curl fails, nothing below will work either - fix connectivity to the proxy first; agentic_ai.gateway's errors will otherwise (correctly) just tell you the same thing: it can't reach http://localhost:4000.

Only once you have a real, running, reachable LiteLLM proxy do the examples below have anything to talk to.

Quickstart: LLM Gateway

1. Connect to it

from agentic_ai.gateway import LiteLLMGateway

# No arguments needed for the common case: connects to
# http://localhost:4000, LiteLLM's own default port.
gateway = LiteLLMGateway()

reply = gateway.complete(
    model="gpt-4o-mini",  # must be registered on your proxy, e.g. in litellm_config.yaml
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "Name three benefits of distributed tracing."},
    ],
)
print(reply)

2. Configuring host, port, and auth

from agentic_ai.gateway import LiteLLMGateway

# Custom port - your proxy isn't on LiteLLM's default 4000
gateway = LiteLLMGateway(port=5001)

# Custom host and port - a proxy running elsewhere on your network
gateway = LiteLLMGateway(host="litellm.internal", port=8080)

# Full base_url - anything host/port can't express (TLS, a path prefix)
gateway = LiteLLMGateway(base_url="https://litellm.example.com/proxy")

# A proxy that requires a virtual key
gateway = LiteLLMGateway(api_key="sk-...")  # resolve this from your own
                                             # secrets store - the gateway
                                             # module doesn't fetch it for you

3. The full response, not just the text

complete() is a convenience wrapper around chat_completion(), which returns the full OpenAI-compatible response body (usage, finish_reason, etc.) when you need more than just the message content:

result = gateway.chat_completion(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Summarize this in one sentence: ..."}],
    temperature=0.2,
    max_tokens=200,
)
print(result["choices"][0]["message"]["content"])
print(result["usage"])

4. Handling errors

The gateway never lets a raw network exception escape - callers get one of two exceptions, so "the proxy is down" and "the proxy rejected the request" are never conflated:

from agentic_ai.gateway import GatewayConnectionError, GatewayRequestError, LiteLLMGateway

gateway = LiteLLMGateway()

try:
    reply = gateway.complete("gpt-4o-mini", [{"role": "user", "content": "hi"}])
except GatewayConnectionError:
    # Nothing is listening at gateway.base_url at all - is LiteLLM
    # actually running? (see Prerequisites above)
    ...
except GatewayRequestError as e:
    # The proxy responded, but with an error (bad model name, missing
    # api_key, malformed request) - e includes the proxy's own message.
    print(e)

5. Cleaning up

LiteLLMGateway holds an open HTTP connection pool; close it when you're done, or use it as a context manager:

with LiteLLMGateway() as gateway:
    reply = gateway.complete("gpt-4o-mini", [{"role": "user", "content": "hi"}])
# connection pool closed automatically here

Evaluation

A complete production evaluation surface for single- and multi-agent AI systems - 48 metrics across 9 categories, organized one folder per category under agentic_ai.evaluation:

Category	Import	Measures
Agent	agentic_ai.evaluation.agent	Task Success/Correctness, Planning, Reasoning, Execution, Recovery, Autonomy, Loops, Lifecycle
LLM	agentic_ai.evaluation.llm	Response Correctness, Groundedness, Hallucination Rate, Instruction Following, Safety/Policy Violation, Latency/Tokens/Cost
Tools	agentic_ai.evaluation.tools	Selection/Argument Accuracy, Success/Failure Rate, Unnecessary Calls, Latency
Multi-Agent	agentic_ai.evaluation.multi_agent	Routing, Delegation, Handoff, Coordination, Duplicate Work
RAG	agentic_ai.evaluation.rag	Recall@K, Context Relevance, Groundedness, Citation Accuracy
Security	agentic_ai.evaluation.security	Prompt Injection, Unauthorized Execution, PII/Cross-Tenant Leakage, Authorization Violations
Platform	agentic_ai.evaluation.platform	Error Rate, Timeout Rate, Cost per Successful Task, SLA Compliance
Memory	agentic_ai.evaluation.memory	Retrieval Accuracy, Consistency
Drift	agentic_ai.evaluation.drift	Statistical (z-score) drift on Success/Correctness/Hallucination/Latency/Cost

Every category is deterministic, LLM-judged, or a documented mix of both - deterministic metrics need no LLM call at all (they read fields you already populated); judged metrics reuse the same LLMJudge from agentic_ai.evaluation.core, built on the gateway above, nothing else.

Deterministic - no LLM call needed

from agentic_ai.evaluation.agent import AgentTrace, compute_task_execution

traces = [
    AgentTrace(run_id="r1", task="find backend jobs", task_succeeded=True),
    AgentTrace(run_id="r2", task="find backend jobs", task_succeeded=False),
    AgentTrace(run_id="r3", task="find backend jobs", task_succeeded=True),
]
metrics = compute_task_execution(traces)
print(metrics.success_rate)  # 0.6666666666666666

LLM-judged - needs a gateway, same one as above

from agentic_ai.evaluation import LLMJudge
from agentic_ai.evaluation.llm import LLMCall, judge_response_correctness
from agentic_ai.gateway import LiteLLMGateway

judge = LLMJudge(LiteLLMGateway(), model="gpt-4o-mini")
call = LLMCall(call_id="c1", model="gpt-4o-mini", prompt="What is 2+2?", response="4")

result = judge_response_correctness(judge, call)
print(result.correct, result.score)

Every judge_*() function across every category takes an optional system_prompt override - the built-in DEFAULT_* rubric is a real, usable starting point, not the only valid one for every domain:

from agentic_ai.evaluation.llm import judge_response_correctness

legal_rubric = "You are a strict legal-domain correctness judge. ..."
result = judge_response_correctness(judge, call, system_prompt=legal_rubric)

Everything at once, persisted, compared over time

Agent Evaluation ties every deterministic + judged category together into one report, storable and diffable:

from agentic_ai.evaluation.agent import evaluate, JSONLEvaluationStore, compare

report = evaluate(traces, judge=judge)  # runs every computable category
store = JSONLEvaluationStore("eval_runs.jsonl")
store.save(report)

baseline = store.list_runs(limit=2)[1]
regressions = compare(baseline, report)  # direction-aware: knows failure_rate up is bad

For statistical drift across many runs over time (not just two points), see agentic_ai.evaluation.drift.compute_drift() and its five named wrappers (compute_task_success_drift, compute_correctness_drift, compute_hallucination_drift, compute_latency_drift, compute_cost_drift).

Every category's own trace/call shape

agent, llm, multi_agent, rag, security, and memory each have their own input model (AgentTrace, LLMCall, MultiAgentTrace, RAGQuery, AuthorizationCheck/TenantDataCheck, MemoryRetrieval) - populate the one your category needs from your own agent's logging; nothing in this library runs an agent or a retriever for you, it only evaluates the record you hand it.

Memory & State

Ten memory types, each a small facade bound to a scope (a session id, an agent id, a namespace) that knows its own purpose and picks sensible defaults - backed by your choice of in-memory, file, SQLite, Redis, Postgres (+pgvector), or Qdrant. No setup needed for local development; pass a URL when you're ready for something durable.

Facade	Import	Backs
SessionMemory	agentic_ai.memory.SessionMemory	Data scoped to one conversation/session
AgentStateMemory	agentic_ai.memory.AgentStateMemory	An agent's own operating state across turns
ShortTermMemory	agentic_ai.memory.ShortTermMemory	Recent context that outlives a single call
WorkingMemory	agentic_ai.memory.WorkingMemory	Scratch space for one in-flight task
LongTermMemory	agentic_ai.memory.LongTermMemory	Durable facts kept across sessions
SemanticMemory	agentic_ai.memory.SemanticMemory	Facts retrieved by meaning (vector search)
EpisodicMemory	agentic_ai.memory.EpisodicMemory	Past events/experiences, recallable by similarity
ProceduralMemory	agentic_ai.memory.ProceduralMemory	Versioned rules/workflows/operating procedures
DocumentMemory	agentic_ai.memory.DocumentMemory	Large source material (PDFs, contracts) + chunk search
SharedMemory	agentic_ai.memory.SharedMemory	A blackboard multiple agents read/write together

Zero-setup quickstart

from agentic_ai.memory import SessionMemory

session = SessionMemory("session-42")  # defaults to an in-process InMemoryStore
session.set("last_intent", "book_flight", ttl_seconds=1800)
print(session.get("last_intent").value)  # "book_flight"

Choosing a backend: just pass a URL

Every key/value facade accepts either an already-constructed store (full control - store=RedisStore(...), or a wrapper-composed one, see below) or a plain shorthand - pick exactly one:

from agentic_ai.memory import SessionMemory, LongTermMemory, AgentStateMemory

SessionMemory("session-42", redis_url="redis://localhost:6379/0")
LongTermMemory("user-123", postgres_url="postgresql://user:pass@localhost:5432/mydb")
AgentStateMemory("agent-7", sqlite_path="agent_state.db")
LongTermMemory("user-123", file_path="longterm.json")  # zero-setup but persisted to disk

redis_url/postgres_url need the matching install extra:

pip install 'enterprise-agentic-ai-framework[redis]'
pip install 'enterprise-agentic-ai-framework[postgres]'
pip install 'enterprise-agentic-ai-framework[qdrant]'
pip install 'enterprise-agentic-ai-framework[memory]'  # all three

Each facade connects and verifies immediately (a real ping / schema init) - a bad URL fails fast in the constructor, not on some later, unrelated call. This library is a client for all of these, never a process manager: deploy Redis/Postgres/Qdrant yourself, same rule as the LLM gateway above.

Semantic, episodic, and document memory (vector-backed)

Embedding generation is always your job - these facades store and search vectors, they never call an embedding model themselves:

from agentic_ai.memory import SemanticMemory

memory = SemanticMemory("user-123", qdrant_url="http://localhost:6333", embedding_dim=1536)
memory.remember("pref-1", "prefers window seats", embedding=embed("prefers window seats"))
results = memory.recall(embed("seating preference"), top_k=3)
print(results[0].record.text, results[0].score)

EpisodicMemory and DocumentMemory combine a plain store (the log / the raw document) with a vector index (similarity recall / chunk search) - pass both explicitly, since a log store and a vector index rarely share connection details:

from agentic_ai.memory import EpisodicMemory
from agentic_ai.memory.stores.sqlite_store import SQLiteStore
from agentic_ai.memory.vector_stores.qdrant_store import QdrantVectorStore

episodes = EpisodicMemory(
    "agent-7",
    store=SQLiteStore("episodes.db"),
    vector_store=QdrantVectorStore(url="http://localhost:6333", embedding_dim=1536),
)
episodes.log_episode("ep-1", "deploy failed: dependency X unavailable", embedding=embed(...))
episodes.recall_similar(embed("deployment failure"), top_k=3)

ProceduralMemory is automatically versioned - every overwrite keeps its prior value retrievable:

from agentic_ai.memory import ProceduralMemory

procedures = ProceduralMemory("support-bot", postgres_url="postgresql://...")
procedures.set("refund_policy", {"max_days": 30, "requires_receipt": True})
procedures.get_history("refund_policy")  # every prior version, oldest first

Governance, security, versioning, checkpointing, audit

Cross-cutting concerns are wrappers that compose onto any store, not sixteen separate storage systems:

from agentic_ai.memory.stores.redis_store import RedisStore
from agentic_ai.memory.wrappers import Actor, GovernedStore, RetentionPolicy, SecureStore
from agentic_ai.memory.core.models import MemoryType

base = RedisStore(url="redis://localhost:6379/0")

governed = GovernedStore(base, RetentionPolicy(
    default_ttl_seconds={MemoryType.SESSION: 1800},
    require_consent=True,
))
governed.set("session-42", "k", "v", memory_type=MemoryType.SESSION, consent=True)

secure = SecureStore(base, enforce_tenant_prefix=True)
actor = Actor(actor_id="u1", roles=["admin"], tenant_id="tenantA")
secure.set(actor, "tenantA:session-42", "k", "v")  # raises AccessDeniedError outside tenantA

checkpoint()/restore() snapshot and roll back a whole scope; VersionedStore (what ProceduralMemory uses internally) keeps a history on every write; AuditedStore emits an event to a sink you provide for every operation, success or failure. All five live in agentic_ai.memory.wrappers and take any MemoryStore - stack as many as you need.

One caveat: keys, not just types, need to be distinct

Every backend keys a record by (scope, key) only - memory_type is stored on the record for filtering, not as part of the write key. Two facades of different types that share both the same scope and the same key on the same store will overwrite each other, same as two dict writes to the same key would. In practice this is rarely an issue - one scope with many distinct keys across several facade types is a normal, safe pattern (list()/clear() on each facade only ever touch its own memory_type).

Context Engineering

Building the actual runtime context for one LLM call - the four pillars (Write, Select, Compress, Isolate) plus one core layer that ties them together: assemble(), the single function that takes whatever candidate context you've gathered and produces a budget- fitted, ordered, cache-boundary-marked result.

Everything in select/compress/isolate operates on plain ContextItems - a small model carrying content, its source (provenance), a priority, an optional relevance score, and a trust level - so every stage composes through the same shape instead of each pillar inventing its own.

Assembly - the core entry point

from agentic_ai.context import assemble, ContextItem, ContextBudget

items = [
    ContextItem(id="sys", section="system", role="system", source="system_prompt",
                content="You are a booking assistant.", priority=1.0, cacheable=True),
    ContextItem(id="turn", section="conversation", source="conversation",
                content="Book me a flight to Denver.", priority=0.95),
]
budget = ContextBudget(max_tokens=4000, reserved_for_output=1000)
result = assemble(items, budget, section_order=["system", "memory", "tools", "conversation"])

print(result.total_tokens)
print([i.id for i in result.excluded])   # what got left out, and why - Context Observability
messages = result.to_messages()          # ready for agentic_ai.gateway.chat_completion(messages=...)

assemble() handles budget packing (drops lowest-priority items first, per-section limits respected), ordering, conflict resolution (two items sharing a metadata["conflict_key"] with different content - newest wins by default), and cache-boundary marking (a cacheable=True item only keeps that flag while it's part of an unbroken cacheable prefix, since prompt caching only pays off on a shared, stable run of leading content). Pass allow_partial=False to raise ContextBudgetExceededError instead of silently dropping anything.

Write - Scratchpad + Memory

Scratchpad is new: ephemeral, ordered notes for one run. "Memories" - the part of Write meant to outlive the run - is agentic_ai.memory itself; nothing here wraps it, there's nothing to add:

from agentic_ai.context.write import Scratchpad

pad = Scratchpad("run-42", redis_url="redis://localhost:6379/0")  # same backend shorthand as agentic_ai.memory
pad.write("tried endpoint A, got a 404")
pad.write("trying endpoint B next")
pad.read_all()               # ordered ScratchpadEntry list
pad.to_context_items()       # ready for assemble()

Select - relevance, prioritization, grounding, freshness, routing

from agentic_ai.memory import SemanticMemory
from agentic_ai.context.select import from_vector_matches, prioritize, require_grounding, filter_stale

memory = SemanticMemory("user-123")
matches = memory.recall(embed("seating preference"), top_k=5)
items = from_vector_matches(matches, section="memory", source="memory:semantic")

grounded, _ = require_grounding(items)              # drops anything with no source
fresh, _ = filter_stale(grounded)                    # drops anything past its ttl_seconds
ranked = prioritize(fresh)                           # combines priority + relevance_score into one ranking

select_by_relevance() also ships a basic, dependency-free keyword scorer for candidates that didn't come from a vector search; route() fans a query out across several named sources (memory, a RAG index, a tools catalog) and merges the results.

Compress - trim, summarize, rolling summary

from agentic_ai.context.compress import trim_to_budget, summarize_items, update_rolling_summary, RollingSummaryState

trimmed = trim_to_budget(candidate_items, max_tokens=2000)           # deterministic, no LLM call

from agentic_ai.gateway import LiteLLMGateway
gateway = LiteLLMGateway()
summary_item = summarize_items(gateway, "gpt-4o-mini", old_turns)     # LLM-based, for when trimming would cut load-bearing info

state = RollingSummaryState()
state = update_rolling_summary(gateway, "gpt-4o-mini", state, all_turns, keep_recent=10)
# keeps the last 10 turns verbatim + a running summary of everything older

Isolate - trust boundaries, tenant scoping, sub-agent partitioning

from agentic_ai.context.isolate import mark_trust_boundary, wrap_untrusted, enforce_tenant_scope, partition_context

marked = mark_trust_boundary(items, trusted_sources={"system_prompt", "memory:semantic"})
safe = [wrap_untrusted(i) for i in marked]   # delimits untrusted (e.g. tool/web) content so it can't pose as an instruction

allowed, _ = enforce_tenant_scope(items, tenant_id="tenantA")

# One assemble() per sub-agent, each with its own budget - one sub-agent's
# clutter never eats another's window:
results = partition_context(
    {"researcher": researcher_items, "writer": writer_items},
    {"researcher": ContextBudget(max_tokens=4000), "writer": ContextBudget(max_tokens=4000)},
)

isolate.security is structural, not a detection engine - it doesn't classify content as an attack (that's the planned guardrails module's job), it enforces what's already known: untrusted content gets delimited, cross-tenant content gets filtered out.

Observability - what context actually went to the model

from agentic_ai.context import ContextTracer

tracer = ContextTracer("run-42", postgres_url="postgresql://...")  # same backend shorthand again
trace = tracer.record(result)   # result from assemble() above
tracer.list_traces()            # every trace recorded for this run, oldest first

Requirements

• Python 3.10+
• A LiteLLM proxy you deploy yourself (this library is a client, not a bundled server)
• For Memory & State: nothing extra for in-memory/file/SQLite; Redis, Postgres (+pgvector), or Qdrant only if you choose those backends

License

Apache-2.0