Aura — aura-context-engine

A framework-agnostic context layer for LLM agents.

Retain every turn durably. Transmit per turn only what the current question needs — under an explicit token budget, with no LLM call on the read path.

Drop it into LangGraph, Agno, Google ADK, or a raw model loop. The engine owns the session; the orchestrator stays an orchestrator.

Why this exists

Long agent transcripts waste three things at once:

• Latency — prefill grows with every unused token
• Cost — you pay for tokens the model does not need
• Attention — models lose the middle of a long prompt (Lost in the Middle)

aura-context-engine keeps the full transcript in storage and assembles a budgeted prompt: pinned facts, the last turn, query-relevant recall, and (for teams) a distilled handoff — never the other agent's raw history.

Install

Python 3.10+.

pip install aura-context-engine

Until the first PyPI release, install from GitHub:

pip install "aura-context-engine @ git+https://github.com/hypen-code/aura.git"

From a local clone (editable):

git clone https://github.com/hypen-code/aura.git
cd aura
python3 -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install -e .

Optional extras

The core has zero required dependencies. Add only what you use.

pip install "aura-context-engine[embed,extract,langgraph]"

30-second usage

from context_engine import ContextEngine, ContextPolicy, PolicyMode

engine = ContextEngine()
session = engine.new_session()

engine.record_exchange(
    session,
    "I live in Oslo and I have two cats, Miso and Nori.",
    "Noted.",
)
engine.record_exchange(session, "I just moved to Berlin.", "Got it.")

policy = ContextPolicy(
    mode=PolicyMode.PRIORITY_SELECTIVE,
    recency_turns=1,
    top_k=4,
    token_budget=2048,
)

messages = engine.build_context(
    session,
    "Where do I live now, and what are my cats called?",
    policy=policy,
    system_prompt="Answer from the conversation only.",
)
# messages is a list of {role, content} dicts — pass it to any LLM API

Run the same flow from the repo:

python examples/quickstart.py

Add it to an orchestrator

The contract is two calls:

1. Write — record_turn / record_exchange after a completed turn
2. Read — build_context (or an adapter hook) before the next model call

Any framework (no adapter)

from context_engine import ContextEngine, ContextPolicy, PolicyMode

engine = ContextEngine()
session = engine.new_session()
policy = ContextPolicy(mode=PolicyMode.PRIORITY_SELECTIVE, token_budget=2048)

def reply(question: str, call_model) -> str:
    prompt = engine.build_context(session, question, policy=policy)
    answer = call_model(prompt)
    engine.record_exchange(session, question, answer)
    return answer

LangGraph

from langgraph.prebuilt import create_react_agent
from context_engine import ContextEngine, ContextPolicy, PolicyMode
from context_engine.adapters import LangGraphContextAdapter

engine = ContextEngine()
adapter = LangGraphContextAdapter(
    engine,
    policy=ContextPolicy(mode=PolicyMode.PRIORITY_SELECTIVE, token_budget=2048),
)
agent = create_react_agent(model=llm, tools=tools, pre_model_hook=adapter.pre_model_hook)

The hook replaces the model input with the assembled context and passes the live tool-call tail through untouched.

Agno

from agno.agent import Agent
from context_engine import ContextEngine
from context_engine.adapters import AgnoContextAdapter

engine = ContextEngine()
adapter = AgnoContextAdapter(engine)
agent = Agent(model=model, add_history_to_context=False)

prompt = adapter.messages_for(question)
response = agent.run(input=adapter.to_agno(prompt))
adapter.record_exchange(question, response.content)

Google ADK

from context_engine import ContextEngine
from context_engine.adapters import GoogleADKContextAdapter

engine = ContextEngine()
adapter = GoogleADKContextAdapter(engine, app_name="my-app")
session = await adapter.prepare_session(runner)
async for event in adapter.run(runner, session, question, agent_name="assistant"):
    ...

Multi-agent teams

Each agent gets its own scoped session. Cross-agent state moves as a handoff packet (short summary + identifier slots), not a shared transcript.

from context_engine import ContextEngine, TeamContext
from context_engine.adapters import LangGraphTeamAdapter

team = TeamContext(ContextEngine(), "order-1")
adapter = LangGraphTeamAdapter(team)

researcher = create_react_agent(
    model, tools, pre_model_hook=adapter.node_hook("researcher")
)
writer = create_react_agent(
    model, tools, pre_model_hook=adapter.node_hook("writer")
)

adapter.record_exchange("researcher", question, findings)
adapter.record_handoff("researcher", "writer", findings)

Same pattern: AgnoTeamAdapter, ADKTeamAdapter.

How assembly works

P1  system prompt + session facts + current question     never evicted
P2  most recent turn(s)                                  last
P3  handoffs / last tool response                        just before the question
P4  top-k query-relevant older turns                     recalled
P5  everything else                                      dropped first

Eviction under token_budget: P5 → shrink P4 → surplus P3 → shrink P2. A single remaining P3 block is never dropped (withholding a handoff is worse than a slightly over-budget prompt).

Write path is async: the turn is stored immediately; fact extraction and embeddings run on a background thread. The newest turn is already in P2, so the next read does not wait.

On any engine error the read path falls back to full-history replay.

Persistence

from context_engine import ContextEngine, SQLiteStorage

engine = ContextEngine(storage=SQLiteStorage("aura.db"))

Default is in-memory. SQLite uses WAL. Restart keeps turns, facts, and undelivered handoffs.

Configuration

ContextPolicy(
    mode=PolicyMode.PRIORITY_SELECTIVE,  # or FULL_HISTORY, WINDOWED, WINDOW_SUMMARY
    recency_turns=2,
    top_k=4,
    token_budget=4096,
    min_relevance=0.15,
    max_facts=12,
    tool_top_k=0,            # 0 = show every tool; >0 admits a subset
    max_result_tokens=0,     # 0 = verbatim tool results
    skill_top_k=0,           # 0 = all skill bodies; >0 manifests only
    min_handoff_relevance=0, # 0 = always display delivered packets
    max_handoff_tokens=256,
)

Optional environment:

Without extract (or if the LLM call fails) the engine uses a heuristic extractor. Without embed it uses BM25.

What this repo is

eval/ and docs/research/ are how the layer was designed and measured. You do not need them to use the package.

Development

git clone https://github.com/hypen-code/aura.git
cd aura
python3 -m venv .venv && source .venv/bin/activate
pip install -e ".[embed,extract]"
python -m unittest discover -s tests -v

See CONTRIBUTING.md.

License

MIT. Contributions are welcome under the same license.

Cite / read more

• Design: docs/research/proposed_solution.md
• Unified assembly (M9): docs/implementation/m9_unified_assembly.md
• Liu et al., Lost in the Middle, arXiv:2307.03172
• Cemri et al., MAST, arXiv:2503.13657
• Zhang et al., ACE, arXiv:2510.04618