ContextForge

Context-native AI orchestration runtime.
Build production LLM applications with explicit, observable, budget-aware control over every token in your context window.

---

The problem with existing frameworks

Most LLM orchestration frameworks treat the context window as a string-concatenation problem. You retrieve some documents, append conversation history, prepend a system prompt, and hope it all fits. When it doesn't, you write ad-hoc truncation logic. When an agent goes deep into a loop, you have no visibility into what's actually inside the window — and no principled way to fix it.

ContextForge is built on a different idea: the context window is a finite, precious resource that deserves its own engineering layer.

Instead of wiring components into pipelines, you declare slots — typed, budget-aware, priority-ranked segments of the context window. The engine handles compression, eviction, versioning, and observability automatically. You stay in control of what goes in, in what order, and what happens when the window fills up.

---

Core concepts

Context Slots

A ContextSlot is a named segment of the context window with a declared token budget, priority, and compression policy.

from contextforge.slots import RetrievedDocSlot, ConversationSlot, SystemSlot

system  = SystemSlot(token_budget=512)         # SYSTEM priority — never evicted
docs    = RetrievedDocSlot(token_budget=3000)  # LOW priority — compressed first
history = ConversationSlot(token_budget=4000)  # MEDIUM priority — top_k compression

Six built-in slot types cover the full production stack:

Slot	Priority	Default Compression	Use for
SystemSlot	SYSTEM	truncate	Persona, identity, top-level rules
GuardrailSlot	CRITICAL	truncate	PII policy, schema constraints
InstructionSlot	HIGH	summarize	Per-request task instructions
MemorySlot	MEDIUM	summarize	Long-term episodic memory
ConversationSlot	MEDIUM	top_k	Chat history
RetrievedDocSlot	LOW	top_k	RAG document chunks
ToolResultSlot	EPHEMERAL	drop	Function call outputs

ForgeEngine

The engine assembles context on every turn:

1. Collects SlotContribution objects from registered providers
2. Prunes expired contributions
3. Applies compression to any slot that exceeds its token_budget
4. Enforces the global token budget by evicting lowest-priority slots
5. Renders the final ordered context string
6. Snapshots a ContextVersion (git-like history)
7. Records a BudgetTracker entry (token heat maps)

from contextforge import ForgeEngine
from contextforge.slots import SystemSlot, RetrievedDocSlot, ConversationSlot

engine = ForgeEngine(total_budget=8192)
engine.register_slot(SystemSlot().add("You are a helpful assistant."))
engine.register_slot(RetrievedDocSlot(token_budget=3000))
engine.register_slot(ConversationSlot(token_budget=4000))

ctx = await engine.assemble(query="What is ContextForge?")
print(ctx.rendered)        # final assembled string, ready for your LLM
print(ctx.total_tokens)    # actual token count
print(ctx.utilization)     # 0.0–1.0 budget utilization

Compression strategies

Four strategies ship out of the box — each slot picks one:

Strategy	Behaviour	When to use
truncate	Keep leading content up to budget	System prompts, guardrails
top_k	Keep highest-scoring contributions (recency + size blend)	Conversation history, retrieved docs
summarize	Secondary LLM call to compress (requires openai extra)	Long-lived memory, instructions
drop	Discard all contributions	Ephemeral tool results
hybrid	TopK then Summarize if still over budget	Production memory slots

Compression is applied automatically when slot.raw_token_estimate > slot.token_budget. You never write truncation logic yourself.

Context versioning

Every assembled context gets a ContextVersion — an immutable snapshot with a content hash. The ContextVersionGraph maintains the full turn history and lets you diff any two versions:

log  = engine.version_log()          # newest-first list of version dicts
diff = engine.context_diff(v1_id, v2_id)  # token delta per slot

This is invaluable for debugging hallucinations in agentic loops — you can replay exactly what was in the model's window at the turn it went wrong.

Token heat maps

BudgetTracker records slot token usage per turn and renders ASCII heat maps for instant visibility into where your budget is going:

  ContextForge Token Heat Map
  ──────────────────────────────────────────────────────
  conversation         ████████████████████████░░░░░░░░░░░░░░░░   4231 tok
  retrieved_docs       ████████████████░░░░░░░░░░░░░░░░░░░░░░░░   2847 tok
  memory               ████░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░    612 tok
  system               █░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░    198 tok
  ──────────────────────────────────────────────────────
  TOTAL                                                           7888 tok

---

Installation

pip install contextforge

With optional LLM provider support:

pip install "contextforge[openai]"      # OpenAI + SummarizeStrategy
pip install "contextforge[anthropic]"   # Anthropic Claude
pip install "contextforge[vector]"      # Qdrant + sentence-transformers
pip install "contextforge[all]"         # Everything

Requires Python 3.10+.

---

Quick start

RAG pipeline

import asyncio
from contextforge import ForgeEngine
from contextforge.slots import SystemSlot, RetrievedDocSlot, ConversationSlot
from contextforge.providers import InMemoryVectorProvider

async def main():
    engine = ForgeEngine(total_budget=8192)
    engine.register_slot(SystemSlot().add("You are a knowledgeable assistant."))
    engine.register_slot(RetrievedDocSlot(token_budget=3000, compression="top_k"))
    engine.register_slot(ConversationSlot(token_budget=4000))

    # Wire a retrieval provider
    docs = InMemoryVectorProvider(top_k=5)
    docs.add_documents(["ContextForge uses slot-based context assembly.", ...])
    engine.register_provider("retrieved_docs", docs)

    ctx = await engine.assemble(query="How does slot compression work?")
    print(ctx.rendered)

asyncio.run(main())

Multi-turn chat with ForgeSession

from contextforge import ForgeEngine, ForgeSession
from contextforge.slots import SystemSlot, ConversationSlot

async def my_llm(messages):
    # Wire to OpenAI / Anthropic / Ollama here
    ...

engine = ForgeEngine(total_budget=8192)
engine.register_slot(SystemSlot().add("You are Forge, a context-aware assistant."))
engine.register_slot(ConversationSlot(token_budget=6000))

session = ForgeSession(engine=engine, llm_fn=my_llm)
response = await session.chat("What can you tell me about context windows?")
session.print_heat_map()

Autonomous agent

from contextforge import ForgeEngine, ForgeAgent
from contextforge.slots import SystemSlot, ToolResultSlot

async def search(query: str) -> str:
    # Your search tool implementation
    ...

engine = ForgeEngine(total_budget=8192)
engine.register_slot(SystemSlot().add("You are a research agent."))
engine.register_slot(ToolResultSlot(token_budget=2048))

agent = ForgeAgent(
    engine=engine,
    llm_fn=my_llm,
    tools={"search": search},
    max_steps=8,
)

result = await agent.run("What are the key differences between RAG and fine-tuning?")
print(result.answer)
print(f"Completed in {result.total_turns} steps, {result.total_tokens} tokens")

Declarative YAML config

# contextforge.yaml
total_budget: 8192
model: gpt-4o
enable_versioning: true
enable_budget_tracking: true

slots:
  - name: system
    type: SystemSlot
    token_budget: 512
  - name: retrieved_docs
    type: RetrievedDocSlot
    token_budget: 3000
    compression: top_k
  - name: conversation
    type: ConversationSlot
    token_budget: 4000
    compression: top_k

slot_order: [system, retrieved_docs, conversation]

from contextforge import ForgeEngine, ForgeConfig

config = ForgeConfig.from_yaml("contextforge.yaml")
engine = ForgeEngine(config=config)

---

Project structure

contextforge/
├── contextforge/
│   ├── engine.py              # ForgeEngine — core assembly runtime
│   ├── slots/
│   │   ├── base.py            # ContextSlot, SlotPriority, SlotContribution
│   │   └── types.py           # SystemSlot, ConversationSlot, RetrievedDocSlot, …
│   ├── compression/
│   │   └── __init__.py        # TruncateStrategy, TopKStrategy, DropStrategy, …
│   ├── providers/
│   │   └── base.py            # BaseProvider, InMemoryVectorProvider, …
│   ├── observability/
│   │   ├── budget_tracker.py  # Token heat maps, per-turn snapshots
│   │   └── context_version.py # ContextVersionGraph, git-like diffing
│   ├── runtime/
│   │   ├── session.py         # ForgeSession — multi-turn chat
│   │   └── agent.py           # ForgeAgent — agentic ReAct loop
│   └── config/
│       └── forge_config.py    # ForgeConfig, YAML loader
├── examples/
│   ├── basic_rag.py
│   ├── multi_turn_chat.py
│   ├── agent_loop.py
│   ├── yaml_config.py
│   └── budget_pressure.py
├── tests/
│   ├── test_slots/
│   ├── test_providers/
│   └── test_runtime/
├── contextforge.yaml          # Default config template
├── pyproject.toml
└── README.md

---

Running examples

# Clone and install in dev mode
git clone https://github.com/your-org/contextforge
cd contextforge
pip install -e ".[dev]"

# Run the examples (no external API key needed for demos)
python examples/basic_rag.py
python examples/multi_turn_chat.py
python examples/agent_loop.py
python examples/yaml_config.py
python examples/budget_pressure.py

---

Running tests

pip install -e ".[dev]"
pytest
pytest --cov=contextforge --cov-report=term-missing

---

How ContextForge differs from existing frameworks

ContextForge is not a pipeline orchestrator. It does not ask you to define a DAG of components, register a document store, or configure a prompt template. Those abstractions exist elsewhere and serve specific purposes.

ContextForge starts from a different premise: the context window is the product. Every other concern — retrieval, memory, tools, history — is just a source of content for a slot. The engine's job is to assemble that content into the best possible window within your budget, automatically, on every single turn.

The result is a layer you can place under any existing pipeline or agent framework, or use as your primary orchestration layer when context fidelity matters more than component reuse.

Key design choices that set it apart:

Context slots are declared, not assembled imperatively. You define the structure of your context window upfront — its types, budgets, priorities, and compression policies — in Python or YAML. The engine resolves everything else. This means your context architecture is auditable, testable, and reproducible.

Compression is a first-class concern. Rather than silently truncating or throwing an error when you exceed the window, ContextForge applies per-slot compression strategies with explicit policies. You choose what gets summarized, what gets dropped, and what must survive at all costs.

Every token has a provenance. SlotContribution objects carry a source_id, token_estimate, and optional expires_at. The engine knows where every token came from and can prune stale content automatically between turns.

Context state is versioned. The ContextVersionGraph gives you a full turn-by-turn history of what was in the model's window, with content hashes and per-slot diffs. This is infrastructure-grade observability, not an afterthought.

---

Roadmap

• Qdrant and Weaviate provider adapters
• OpenTelemetry span export from BudgetTracker
• FastAPI observability dashboard (token heat map as a live web UI)
• tiktoken-based accurate token counting
• Multi-agent context handoff (pass a ContextVersion between agents)
• Async streaming context assembly
• LangChain and LlamaIndex bridge adapters

---

Contributing

Contributions welcome. Open an issue to discuss before submitting a large PR.

pip install -e ".[dev]"
ruff check contextforge/
mypy contextforge/
pytest

---

License

Apache 2.0 — see LICENSE.