Corpus OS

<img width="1128" height="191" alt="image" src="https://github.com/user-attachments/assets/cb2fe4ef-be6a-4406-b899-23ad1ed30c08" />

Reference implementation of the Corpus OS — a wire-first, vendor-neutral SDK for interoperable AI frameworks and data backends across four domains: LLM, Embedding, Vector, and Graph.

---

Contact: team@corpusos.com Website: https://corpusos.com Docs: https://docs.corpusos.com

---

┌──────────────────────────────────────────────────────────────────────┐
│  Your App / Agents / RAG Pipelines                                   │
│  (LangChain · LlamaIndex · Semantic Kernel · CrewAI · AutoGen · MCP) │
├──────────────────────────────────────────────────────────────────────┤
│  Corpus OS Protocol and SDK.                                         │
│  One protocol · One error taxonomy · One metrics model               │
├──────────┬──────────────┬────────────┬───────────────────────────────┤
│ LLM/v1   │ Embedding/v1 │ Vector/v1  │ Graph/v1                      │
├──────────┴──────────────┴────────────┴───────────────────────────────┤
│  Any Provider: OpenAI · Anthropic · Pinecone · Neo4j · ...           │
└──────────────────────────────────────────────────────────────────────┘

Keep your frameworks. Standardize your infrastructure.

Open-Core Model: The Corpus OS Protocol Suite and SDK are fully open source (Apache-2.0). Corpus Router and official production adapters are commercial, optional, and built on the same public protocols. Using this SDK does not lock you into CORPUS Router.

---

Table of Contents

1. Why CORPUS
2. How CORPUS Compares
3. When Not to Use CORPUS
4. Install
5. Quick Start
6. Domain Examples
7. Core Concepts
8. Error Taxonomy & Observability
9. Performance & Configuration
10. Testing & Conformance
11. Documentation Layout
12. FAQ
13. Contributing
14. License & Commercial Options

---

Why Corpus OS

Modern AI platforms juggle multiple LLM, embedding, vector, and graph backends. Each vendor ships unique APIs, error schemes, rate limits, and capabilities — making cross-provider integration brittle and costly.

The problem:

• Provider proliferation — Dozens of incompatible APIs across AI infrastructure
• Duplicate integration — Different error handling, observability, and resilience patterns rewritten per provider and framework
• Vendor lock-in — Applications tightly coupled to specific backend choices
• Operational complexity — Inconsistent monitoring and debugging across services

Corpus OS provides:

• Stable, runtime-checkable protocols across all four domains
• Normalized errors with retry hints and machine-actionable scopes
• SIEM-safe metrics (low-cardinality, tenant-hashed, no PII)
• Deadline propagation for cancellation and cost control
• Two modes — compose under your own router (thin) or use lightweight built-in infra (standalone)
• Wire-first design — canonical JSON envelopes implementable in any language, with this SDK as reference

Corpus OS is not a replacement for LangChain, LlamaIndex, Semantic Kernel, CrewAI, AutoGen, or MCP. Use those for agent-specific orchestration, agents, tools, and RAG pipelines. Use Corpus OS to standardize the infrastructure layer underneath them. Your app teams keep their frameworks. Your platform team gets one protocol, one error taxonomy, and one observability model across everything.

---

How Corpus OS Compares

| Aspect | LangChain / LlamaIndex | OpenRouter | MCP | Corpus OS | |---|---|---|---|---| | Scope | Application framework | LLM unification | Tools & data sources | AI infrastructure protocols | | Domains | LLM + Tools | LLM only | Tools + Data | LLM + Vector + Graph + Embedding | | Error Standardization | Partial | Limited | N/A | Comprehensive taxonomy | | Multi-Provider Routing | Basic | Managed service | N/A | Protocol for any router | | Observability | Basic | Limited | N/A | Built-in metrics + tracing | | Vendor Neutrality | High | Service-dependent | High | Protocol-first, no lock-in |

Who is this for?

• App developers — Keep using your framework of choice. Talk to all backends through Corpus OS protocols. Swap providers or frameworks without rewriting integration code.
• Framework maintainers — Implement one CORPUS adapter per protocol. Instantly support any conformant backend.
• Backend vendors — Implement llm/v1, embedding/v1, vector/v1, or graph/v1 once, run the conformance suite, and your service works with every framework.
• Platform / infra teams — Unified observability: normalized error codes, deadlines, and metrics. One set of dashboards and SLOs across all AI traffic.
• MCP users — The Corpus OS MCP server exposes protocols as standard MCP tools. Any MCP client can call into your infra with consistent behavior.

Integration Patterns

| Pattern | How It Works | What You Get | |---|---|---| | Framework → Corpus OS → Providers | Framework uses Corpus OS as client | Unified errors/metrics across providers | | Corpus OS → Framework-as-adapter → Providers | Framework wrapped as Corpus OS adapter | Reuse existing chains/indices as "providers" | | Mixed | Both of the above | Gradual migration, no big-bang rewrites |

Large teams typically run all three patterns at once.

---

When Not to Use CORPUS

You probably don't need Corpus OS if:

• Single-provider and happy — One backend, fine with their SDK and breaking changes.
• No governance pressure — No per-tenant isolation, budgets, audit trails, or data residency.
• No cross-domain orchestration — Not coordinating LLM + Vector + Graph + Embedding together.
• Quick throwaway prototype — Lock-in, metrics, and resilience aren't worth thinking about yet.

If any of these stop being true, corpus_sdk is the incremental next step.

---

Install

pip install corpus_sdk

Python ≥ 3.10 recommended. No heavy runtime dependencies.

---

⚡ Quick Start

import asyncio
from corpus_sdk.llm.llm_base import (
    BaseLLMAdapter, OperationContext, LLMCompletion,
    LLMCapabilities, TokenUsage
)

class QuickAdapter(BaseLLMAdapter):
    async def _do_capabilities(self) -> LLMCapabilities:
        return LLMCapabilities(
            server="quick-demo",
            version="1.0.0",
            model_family="demo",
            max_context_length=4096,
        )
    
    async def _do_complete(self, messages, model=None, **kwargs) -> LLMCompletion:
        return LLMCompletion(
            text="Hello from CORPUS!",
            model=model or "quick-demo",
            model_family="demo",
            usage=TokenUsage(prompt_tokens=2, completion_tokens=3, total_tokens=5),
            finish_reason="stop",
        )
    
    async def _do_count_tokens(self, text, *, model=None, ctx=None) -> int:
        return len(text.split())  # Simple word count
    
    async def _do_health(self, *, ctx=None) -> dict:
        return {"ok": True, "server": "quick-demo"}

# Usage
async def main():
    print("=" * 80)
    print("Quick LLM Adapter Demo")
    print("=" * 80)
    
    adapter = QuickAdapter()
    ctx = OperationContext(request_id="test-123")
    
    # Test 1: Capabilities
    caps = await adapter.capabilities()
    print(f"\n✅ Capabilities:")
    print(f"   Server: {caps.server} v{caps.version}")
    print(f"   Model family: {caps.model_family}")
    print(f"   Max context: {caps.max_context_length}")
    
    # Test 2: Completion
    result = await adapter.complete(
        messages=[{"role": "user", "content": "Hi"}], 
        ctx=ctx
    )
    print(f"\n✅ Completion:")
    print(f"   Response: {result.text}")
    print(f"   Model: {result.model}")
    print(f"   Tokens used: {result.usage.total_tokens} (prompt: {result.usage.prompt_tokens}, completion: {result.usage.completion_tokens})")
    print(f"   Finish reason: {result.finish_reason}")
    
    # Test 3: Token counting
    tokens = await adapter.count_tokens("This is a test message")
    print(f"\n✅ Token Counting:")
    print(f"   Text: 'This is a test message'")
    print(f"   Tokens: {tokens}")
    
    # Test 4: Health check
    health = await adapter.health()
    print(f"\n✅ Health Check:")
    print(f"   OK: {health.get('ok', False)}")
    print(f"   Server: {health.get('server', 'unknown')}")
    
    print("\n" + "=" * 80)
    print("✅ All tests passed!")
    print("=" * 80)

if __name__ == "__main__":
    asyncio.run(main())

A complete quick start with all four protocols is in docs/guides/QUICK_START.md.

---

Domain Examples

Minimal viable adapter: Implement _do_capabilities, your core operation (_do_embed, _do_complete, _do_query, etc.), and _do_health. All other methods have safe no-op defa