corpusos

# 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](https://corpusos.com)** — 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](mailto:team@corpusos.com) **Website:** [https://corpusos.com](https://corpusos.com) **Docs:** [https://docs.corpusos.com](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](https://corpusos.com)** 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](#why-corpus) 2. [How CORPUS Compares](#how-corpus-compares) 3. [When Not to Use CORPUS](#when-not-to-use-corpus) 4. [Install](#install) 5. [Quick Start](#-quick-start) 6. [Domain Examples](#domain-examples) 7. [Core Concepts](#core-concepts) 8. [Error Taxonomy & Observability](#error-taxonomy--observability) 9. [Performance & Configuration](#performance--configuration) 10. [Testing & Conformance](#testing--conformance) 11. [Documentation Layout](#-documentation-layout) 12. [FAQ](#faq) 13. [Contributing](#contributing) 14. [License & Commercial Options](#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 ```bash pip install corpus_sdk ``` Python ≥ 3.10 recommended. No heavy runtime dependencies. --- ## ⚡ Quick Start ```python 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`](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 defaults — you only override what you need. > In all examples, swap `Example*Adapter` with your actual adapter class that inherits the corresponding base and implements `_do_*` hooks. <details> <summary><strong>Embeddings</strong></summary> ```python import asyncio from corpus_sdk.embedding.embedding_base import ( BaseEmbeddingAdapter, EmbedSpec, OperationContext, EmbeddingVector, EmbeddingCapabilities, EmbedResult ) class QuickEmbeddingAdapter(BaseEmbeddingAdapter): async def _do_capabilities(self) -> EmbeddingCapabilities: return EmbeddingCapabilities( server="quick-embeddings", version="1.0.0", supported_models=("quick-embed-001",), max_batch_size=128, max_text_length=8192, supports_normalization=True, normalizes_at_source=False, supports_deadline=True, supports_token_counting=False, ) async def _do_embed(self, spec: EmbedSpec, *, ctx=None) -> EmbedResult: vec = [0.1, 0.2, 0.3] return EmbedResult( embedding=EmbeddingVector( vector=vec, text=spec.text, model=spec.model, dimensions=len(vec) ), model=spec.model, text=spec.text, tokens_used=None, truncated=False, ) async def _do_health(self, *, ctx=None) -> dict: return {"ok": True, "server": "quick-embeddings", "version": "1.0.0"} # Usage async def main(): print("=" * 80) print("Quick Embedding Adapter Demo") print("=" * 80) async with QuickEmbeddingAdapter() as adapter: ctx = OperationContext(request_id="req-1", tenant="acme") # Test 1: Capabilities caps = await adapter.capabilities() print(f"\n✅ Capabilities:") print(f" Server: {caps.server} v{caps.version}") print(f" Supported models: {caps.supported_models}") print(f" Max batch size: {caps.max_batch_size}") print(f" Max text length: {caps.max_text_length}") print(f" Supports normalization: {caps.supports_normalization}") print(f" Supports deadline: {caps.supports_deadline}") # Test 2: Embedding res = await adapter.embed( EmbedSpec(text="hello world", model="quick-embed-001"), ctx=ctx ) print(f"\n✅ Embedding:") print(f" Text: '{res.text}'") print(f" Vector: {res.embedding.vector}") print(f" Dimensions: {res.embedding.dimensions}") print(f" Model: {res.model}") print(f" Truncated: {res.truncated}") # Test 3: Multiple embeddings texts = ["first text", "second text", "third text"] print(f"\n✅ Multiple Embeddings:") for i, text in enumerate(texts, 1): res = await adapter.embed( EmbedSpec(text=text, model="quick-embed-001"), ctx=ctx ) print(f" {i}. '{text}' → {res.embedding.dimensions}D vector") # 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')} v{health.get('version', 'unknown')}") print("\n" + "=" * 80) print("✅ All tests passed!") print("=" * 80) if __name__ == "__main__": asyncio.run(main()) ``` </details> <details> <summary><strong>LLM</strong></summary> ```python import asyncio from corpus_sdk.llm.llm_base import ( BaseLLMAdapter, OperationContext, LLMCompletion, TokenUsage, LLMCapabilities ) class QuickLLMAdapter(BaseLLMAdapter): async def _do_capabilities(self) -> LLMCapabilities: return LLMCapabilities( server="quick-llm", version="1.0.0", model_family="gpt-4", max_context_length=8192, supports_streaming=True, supports_roles=True, supports_json_output=False, supports_parallel_tool_calls=False, idempotent_writes=False, supports_multi_tenant=True, supports_system_message=True, ) async def _do_complete(self, messages, model, **kwargs) -> LLMCompletion: return LLMCompletion( text="Hello from quick-llm!", model=model, model_family="gpt-4", usage=TokenUsage(prompt_tokens=5, completion_tokens=5, total_tokens=10), 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-llm", "version": "1.0.0"} # Usage async def main(): print("=" * 80) print("Quick LLM Adapter Demo") print("=" * 80) async with QuickLLMAdapter() as adapter: ctx = OperationContext(request_id="req-2", tenant="acme") # 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 length: {caps.max_context_length}") print(f" Supports streaming: {caps.supports_streaming}") print(f" Supports roles: {caps.supports_roles}") print(f" Supports system message: {caps.supports_system_message}") print(f" Supports multi-tenant: {caps.supports_multi_tenant}") # Test 2: Completion resp = await adapter.complete( messages=[{"role": "user", "content": "Say hi"}], model="quick-llm-001", ctx=ctx, ) print(f"\n✅ Completion:") print(f" Response: {resp.text}") print(f" Model: {resp.model}") print(f" Model family: {resp.model_family}") print(f" Tokens: {resp.usage.total_tokens} (prompt: {resp.usage.prompt_tokens}, completion: {resp.usage.completion_tokens})") print(f" Finish reason: {resp.finish_reason}") # Test 3: Multi-message conversation messages = [ {"role": "system", "content": "You are a helpful assistant"}, {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there!"}, {"role": "user", "content": "How are you?"} ] resp = await adapter.complete(messages=messages, model="quick-llm-001", ctx=ctx) print(f"\n✅ Multi-turn Conversation:") print(f" Messages: {len(messages)} turns") print(f" Response: {resp.text}") # Test 4: Token counting test_text = "This is a longer text to count tokens for testing purposes" tokens = await adapter.count_tokens(test_text, model="quick-llm-001") print(f"\n✅ Token Counting:") print(f" Text: '{test_text}'") print(f" Tokens: {tokens}") # Test 5: 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')} v{health.get('version', 'unknown')}") print("\n" + "=" * 80) print("✅ All tests passed!") print("=" * 80) if __name__ == "__main__": asyncio.run(main()) ``` </details> <details> <summary><strong>Vector</strong></summary> ```python import asyncio from corpus_sdk.vector.vector_base import ( BaseVectorAdapter, VectorCapabilities, QuerySpec, UpsertSpec, UpsertResult, QueryResult, Vector, VectorMatch, OperationContext, VectorID ) class QuickVectorAdapter(BaseVectorAdapter): def __init__(self): super().__init__() # Simple in-memory storage self.vectors = {} async def _do_capabilities(self) -> VectorCapabilities: return VectorCapabilities( server="quick-vector", version="1.0.0", max_dimensions=3 ) async def _do_upsert(self, spec: UpsertSpec, *, ctx=None) -> UpsertResult: """Store vectors in memory""" ns = spec.namespace or "default" if ns not in self.vectors: self.vectors[ns] = [] self.vectors[ns].extend(spec.vectors) return UpsertResult( upserted_count=len(spec.vectors), failed_count=0, failures=[] ) async def _do_query(self, spec: QuerySpec, *, ctx=None) -> QueryResult: """Search for similar vectors""" ns = spec.namespace or "default" stored_vectors = self.vectors.get(ns, []) # Simple cosine similarity def cosine_sim(a, b): dot = sum(x * y for x, y in zip(a, b)) mag_a = sum(x * x for x in a) ** 0.5 mag_b = sum(x * x for x in b) ** 0.5 return dot / (mag_a * mag_b) if mag_a and mag_b else 0 matches = [] for vec in stored_vectors: score = cosine_sim(spec.vector, vec.vector) matches.append(VectorMatch(vector=vec, score=score, distance=1-score)) # Sort by score descending matches.sort(key=lambda m: m.score, reverse=True) top_matches = matches[:spec.top_k] if spec.top_k else matches return QueryResult( matches=top_matches, query_vector=spec.vector, namespace=ns, total_matches=len(top_matches), ) async def _do_health(self, *, ctx=None) -> dict: return {"ok": True, "server": "quick-vector", "version": "1.0.0"} # Usage - Complete flow async def main(): print("=" * 80) print("Quick Vector Adapter Demo") print("=" * 80) adapter = QuickVectorAdapter() ctx = OperationContext(request_id="req-3", tenant="acme") # Test 1: Capabilities caps = await adapter.capabilities() print(f"\n✅ Capabilities:") print(f" Server: {caps.server} v{caps.version}") print(f" Max dimensions: {caps.max_dimensions}") # Test 2: Upsert vectors vectors_to_add = [ Vector(id=VectorID("v1"), vector=[0.1, 0.2, 0.3], metadata={"label": "first"}), Vector(id=VectorID("v2"), vector=[0.4, 0.5, 0.6], metadata={"label": "second"}), Vector(id=VectorID("v3"), vector=[0.7, 0.8, 0.9], metadata={"label": "third"}), ] upsert_result = await adapter.upsert( UpsertSpec(vectors=vectors_to_add), ctx=ctx ) print(f"\n✅ Upsert:") print(f" Upserted: {upsert_result.upserted_count} vectors") print(f" Failed: {upsert_result.failed_count} vectors") # Test 3: Query for similar vectors (top_k=2 REQUIRED by SDK) query_result = await adapter.query( QuerySpec(vector=[0.1, 0.2, 0.3], top_k=2), ctx=ctx ) print(f"\n✅ Query (top 2):") print(f" Query vector: {query_result.query_vector}") print(f" Total matches: {query_result.total_matches}") print(f" Results:") for i, match in enumerate(query_result.matches, 1): print(f" {i}. ID: {match.vector.id}, Score: {match.score:.3f}, Distance: {match.distance:.3f}") print(f" Metadata: {match.vector.metadata}") # Test 4: Query all vectors (with top_k=10) query_result_all = await adapter.query( QuerySpec(vector=[0.5, 0.5, 0.5], top_k=10), ctx=ctx ) print(f"\n✅ Query (all matches):") print(f" Query vector: [0.5, 0.5, 0.5]") print(f" Total matches: {query_result_all.total_matches}") for i, match in enumerate(query_result_all.matches, 1): print(f" {i}. ID: {match.vector.id}, Score: {match.score:.3f}") # Test 5: Namespace support ns_vectors = [ Vector(id=VectorID("ns1"), vector=[0.9, 0.8, 0.7], metadata={"type": "namespace_test"}), ] upsert_ns = await adapter.upsert( UpsertSpec(vectors=ns_vectors, namespace="test-namespace"), ctx=ctx ) print(f"\n✅ Namespace Support:") print(f" Upserted {upsert_ns.upserted_count} vector(s) to 'test-namespace'") query_ns = await adapter.query( QuerySpec(vector=[0.9, 0.8, 0.7], top_k=5, namespace="test-namespace"), ctx=ctx ) print(f" Query in 'test-namespace': {query_ns.total_matches} match(es)") # Test 6: 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')} v{health.get('version', 'unknown')}") print("\n" + "=" * 80) print("✅ All tests passed!") print("=" * 80) if __name__ == "__main__": asyncio.run(main()) ``` </details> <details> <summary><strong>Graph</strong></summary> ```python import asyncio from corpus_sdk.graph.graph_base import ( BaseGraphAdapter, GraphCapabilities, UpsertNodesSpec, Node, GraphID, OperationContext, GraphQuerySpec, QueryResult ) class QuickGraphAdapter(BaseGraphAdapter): def __init__(self): super().__init__() # Simple in-memory storage self.nodes = {} async def _do_capabilities(self) -> GraphCapabilities: return GraphCapabilities( server="quick-graph", version="1.0.0", supported_query_dialects=("cypher",), supports_stream_query=True, supports_bulk_vertices=True, supports_batch=True, supports_schema=True, ) async def _do_upsert_nodes(self, spec: UpsertNodesSpec, *, ctx=None): # Store nodes in memory for node in spec.nodes: self.nodes[str(node.id)] = node from corpus_sdk.graph.graph_base import UpsertResult return UpsertResult( upserted_count=len(spec.nodes), failed_count=0, failures=[] ) async def _do_query(self, spec: GraphQuerySpec, *, ctx=None) -> QueryResult: return QueryResult( records=[{"id": 1, "name": "Ada"}], summary={"rows": 1}, dialect=spec.dialect, namespace=spec.namespace or "default", ) async def _do_health(self, *, ctx=None) -> dict: return {"ok": True, "server": "quick-graph", "version": "1.0.0"} # Usage async def main(): print("=" * 80) print("Quick Graph Adapter Demo") print("=" * 80) async with QuickGraphAdapter() as adapter: ctx = OperationContext(request_id="req-4", tenant="acme") # Test 1: Capabilities caps = await adapter.capabilities() print(f"\n✅ Capabilities:") print(f" Server: {caps.server} v{caps.version}") print(f" Supported dialects: {caps.supported_query_dialects}") print(f" Supports streaming: {caps.supports_stream_query}") print(f" Supports bulk vertices: {caps.supports_bulk_vertices}") print(f" Supports batch: {caps.supports_batch}") print(f" Supports schema: {caps.supports_schema}") # Test 2: Upsert nodes result = await adapter.upsert_nodes( UpsertNodesSpec(nodes=[ Node( id=GraphID("user:1"), labels=("User",), properties={"name": "Ada"} ), Node( id=GraphID("user:2"), labels=("User",), properties={"name": "Bob"} ) ]) ) print(f"\n✅ Upsert Nodes:") print(f" Upserted: {result.upserted_count} nodes") print(f" Failed: {result.failed_count} nodes") # Test 3: Query the graph query_result = await adapter.query( GraphQuerySpec( text="MATCH (u:User) RETURN u.name", dialect="cypher" ) ) print(f"\n✅ Query:") print(f" Query: MATCH (u:User) RETURN u.name") print(f" Dialect: {query_result.dialect}") print(f" Records: {query_result.records}") print(f" Summary: {query_result.summary}") # Test 4: Check stored nodes print(f"\n✅ Storage Check:") print(f" Total nodes in memory: {len(adapter.nodes)}") for node_id, node in adapter.nodes.items(): print(f" - {node_id}: labels={node.labels}, properties={node.properties}") # Test 5: 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')} v{health.get('version', 'unknown')}") print("\n" + "=" * 80) print("✅ All tests passed!") print("=" * 80) if __name__ == "__main__": asyncio.run(main()) ``` </details> <details> <summary><strong> RAG flow </strong></summary> ```python """ RAG (Retrieval-Augmented Generation) Example Demonstrates combining Embedding, Vector, and LLM protocols """ import asyncio from corpus_sdk.llm.llm_base import ( BaseLLMAdapter, OperationContext, LLMCompletion, TokenUsage, LLMCapabilities ) from corpus_sdk.embedding.embedding_base import ( BaseEmbeddingAdapter, EmbedSpec, EmbeddingVector, EmbeddingCapabilities, EmbedResult ) from corpus_sdk.vector.vector_base import ( BaseVectorAdapter, VectorCapabilities, QuerySpec, UpsertSpec, UpsertResult, QueryResult, Vector, VectorMatch, VectorID ) # 1. Embedding Adapter class QuickEmbeddingAdapter(BaseEmbeddingAdapter): async def _do_capabilities(self) -> EmbeddingCapabilities: return EmbeddingCapabilities( server="quick-embeddings", version="1.0.0", supported_models=("quick-embed-001",), max_batch_size=128, max_text_length=8192, ) async def _do_embed(self, spec: EmbedSpec, *, ctx=None) -> EmbedResult: # Deterministic embedding based on text content vec = [hash(spec.text + str(i)) % 1000 / 1000.0 for i in range(384)] return EmbedResult( embedding=EmbeddingVector( vector=vec, text=spec.text, model=spec.model, dimensions=len(vec) ), model=spec.model, text=spec.text, tokens_used=len(spec.text.split()), truncated=False, ) async def _do_health(self, *, ctx=None) -> dict: return {"ok": True, "server": "quick-embeddings"} # 2. Vector Store Adapter class QuickVectorAdapter(BaseVectorAdapter): def __init__(self): super().__init__() self.vectors = {} async def _do_capabilities(self) -> VectorCapabilities: return VectorCapabilities( server="quick-vector", version="1.0.0", max_dimensions=384 ) async def _do_upsert(self, spec: UpsertSpec, *, ctx=None) -> UpsertResult: ns = spec.namespace or "default" if ns not in self.vectors: self.vectors[ns] = [] self.vectors[ns].extend(spec.vectors) return UpsertResult( upserted_count=len(spec.vectors), failed_count=0, failures=[] ) async def _do_query(self, spec: QuerySpec, *, ctx=None) -> QueryResult: ns = spec.namespace or "default" stored = self.vectors.get(ns, []) # Cosine similarity def cosine_sim(a, b): dot = sum(x * y for x, y in zip(a, b)) mag_a = sum(x * x for x in a) ** 0.5 mag_b = sum(x * x for x in b) ** 0.5 return dot / (mag_a * mag_b) if mag_a and mag_b else 0 matches = [] for vec in stored: score = cosine_sim(spec.vector, vec.vector) matches.append(VectorMatch(vector=vec, score=score, distance=1-score)) matches.sort(key=lambda m: m.score, reverse=True) top_k = matches[:spec.top_k] if spec.top_k else matches return QueryResult( matches=top_k, query_vector=spec.vector, namespace=ns, total_matches=len(top_k), ) async def _do_health(self, *, ctx=None) -> dict: return {"ok": True, "server": "quick-vector"} # 3. LLM Adapter class QuickLLMAdapter(BaseLLMAdapter): async def _do_capabilities(self) -> LLMCapabilities: return LLMCapabilities( server="quick-llm", version="1.0.0", model_family="gpt-4", max_context_length=8192, ) async def _do_complete(self, messages, model, **kwargs) -> LLMCompletion: # Extract the last user message user_msg = messages[-1]["content"] if messages else "" # Simple mock: generate response based on context presence if "Corpus SDK" in user_msg: response = "Based on the documentation, Corpus SDK is a protocol suite that provides standardized interfaces for LLM, Embedding, Vector, and Graph operations. It enables backend-agnostic AI applications." elif "domains" in user_msg.lower(): response = "The SDK supports four core domains: LLM (language models), Embedding (text vectorization), Vector (similarity search), and Graph (knowledge graphs)." elif "integrate" in user_msg.lower() or "backend" in user_msg.lower(): response = "To integrate a backend, create an adapter class that inherits from the appropriate base (BaseLLMAdapter, BaseEmbeddingAdapter, etc.) and implement the _do_* hook methods like _do_complete, _do_embed, or _do_query." else: response = "I can provide information about Corpus SDK based on the available documentation." return LLMCompletion( text=response, model=model, model_family="gpt-4", usage=TokenUsage( prompt_tokens=len(user_msg.split()), completion_tokens=len(response.split()), total_tokens=len(user_msg.split()) + len(response.split()) ), finish_reason="stop", ) async def _do_count_tokens(self, text, *, model=None, ctx=None) -> int: return len(text.split()) async def _do_health(self, *, ctx=None) -> dict: return {"ok": True, "server": "quick-llm"} # 4. RAG Pipeline class RAGPipeline: """Combines embedding, vector search, and LLM for RAG""" def __init__(self): self.embedder = QuickEmbeddingAdapter() self.vector_db = QuickVectorAdapter() self.llm = QuickLLMAdapter() self.ctx = OperationContext(request_id="rag-demo", tenant="quickstart") async def index_documents(self, documents: list[str]) -> int: """Index documents into the vector database""" vectors = [] for i, doc in enumerate(documents): # Embed the document embed_result = await self.embedder.embed( EmbedSpec(text=doc, model="quick-embed-001"), ctx=self.ctx ) # Create vector with metadata vectors.append(Vector( id=VectorID(f"doc-{i}"), vector=embed_result.embedding.vector, metadata={"text": doc, "doc_id": i} )) # Upsert to vector store result = await self.vector_db.upsert( UpsertSpec(vectors=vectors), ctx=self.ctx ) return result.upserted_count async def query(self, question: str, top_k: int = 3) -> dict: """Answer a question using RAG""" # Step 1: Embed the question question_embed = await self.embedder.embed( EmbedSpec(text=question, model="quick-embed-001"), ctx=self.ctx ) # Step 2: Search for relevant documents search_results = await self.vector_db.query( QuerySpec(vector=question_embed.embedding.vector, top_k=top_k), ctx=self.ctx ) # Step 3: Build context from top matches context_docs = [ match.vector.metadata["text"] for match in search_results.matches ] context = "\n\n".join(f"[{i+1}] {doc}" for i, doc in enumerate(context_docs)) # Step 4: Generate answer with LLM prompt = f"""Use the following context to answer the question. Context: {context} Question: {question} Answer:""" completion = await self.llm.complete( messages=[{"role": "user", "content": prompt}], model="quick-llm-001", ctx=self.ctx ) return { "answer": completion.text, "sources": context_docs, "relevance_scores": [m.score for m in search_results.matches], "tokens_used": completion.usage.total_tokens, } # Usage Example async def main(): print("=" * 70) print("RAG Pipeline Demo - Corpus SDK") print("=" * 70) # Initialize pipeline rag = RAGPipeline() # Knowledge base documents documents = [ "Corpus SDK is a protocol suite for building LLM applications with standardized interfaces.", "The SDK supports four core domains: LLM, Embedding, Vector, and Graph protocols.", "Adapters implement _do_* hooks (_do_complete, _do_embed, _do_query, etc.) to integrate any backend.", "All SDK examples are tested and production-ready, ensuring reliability.", "The protocol-based design enables swapping backends without changing application code.", ] # Index documents print("\n📚 Indexing documents...") count = await rag.index_documents(documents) print(f"✅ Indexed {count} documents\n") # Ask questions questions = [ "What is Corpus SDK?", "How many domains does the SDK support?", "How do I integrate my backend?", ] for i, question in enumerate(questions, 1): print(f"\n{'─' * 70}") print(f"Question {i}: {question}") print('─' * 70) result = await rag.query(question, top_k=2) print(f"\n💡 Answer:\n{result['answer']}\n") print(f"📊 Relevance Scores: {[f'{s:.3f}' for s in result['relevance_scores']]}") print(f"🔢 Tokens Used: {result['tokens_used']}") print(f"\n📎 Sources:") for j, source in enumerate(result['sources'], 1): print(f" [{j}] {source}") print("\n" + "=" * 70) print("✅ RAG Demo Complete!") print("=" * 70) if __name__ == "__main__": asyncio.run(main()) ``` </details> Full implementations with batch operations, streaming, and multi-cloud scenarios are in [`docs/guides/ADAPTER_RECIPES.md`](docs/guides/ADAPTER_RECIPES.md). --- ## Core Concepts - **Protocol vs Base** — Protocols define required behavior. Bases implement validation, deadlines, observability, and error normalization. You implement `_do_*` hooks. - **OperationContext** — Carries `request_id`, `idempotency_key`, `deadline_ms`, `traceparent`, `tenant`, and cache hints across all operations. - **Wire Protocol** — Canonical envelopes (`op`, `ctx`, `args`) and response shapes (`ok`, `code`, `result`) defined in [`docs/spec/PROTOCOL.md`](docs/spec/PROTOCOL.md). - **Corpus OS-Compatible** — Implementations that honor the envelopes, reserved `op` strings, and error taxonomy. Validated by the conformance suite. --- ## Error Taxonomy & Observability All domains share a **normalized error taxonomy**: `BadRequest`, `AuthError`, `ResourceExhausted`, `TransientNetwork`, `Unavailable`, `NotSupported`, `DeadlineExceeded`, plus domain-specific errors like `TextTooLong`, `ModelOverloaded`, `DimensionMismatch`, and `IndexNotReady`. Errors carry **machine-actionable hints** (`retry_after_ms`, `throttle_scope`) so routers and control planes can react consistently across providers. A pluggable `MetricsSink` protocol lets you bring your own metrics backend. Bases emit one `observe` per operation, hash tenants before recording, and never log prompts, vectors, or raw tenant IDs. Full details in [`docs/spec/ERRORS.md`](docs/spec/ERRORS.md) and [`docs/spec/METRICS.md`](docs/spec/METRICS.md). --- ## Performance & Configuration Base overhead is typically **<10 ms** relative to vendor SDK calls: validation <1 ms, metrics <0.1 ms, cache lookup (standalone) <0.5 ms. Async-first design avoids blocking and supports high concurrency. Batch operations (`embed_batch`, vector upserts, graph batch) are preferred for throughput. Benchmarks and deployment patterns in [`docs/guides/IMPLEMENTATION.md`](docs/guides/IMPLEMENTATION.md). ### Modes: `thin` vs `standalone` Once you're ready for production, choose a mode: | Mode | Infra Hooks | When to Use | |---|---|---| | **`thin`** (default) | All no-ops | You have an external control plane (router, scheduler, limiter) | | **`standalone`** | Deadline enforcement, circuit breaker, token-bucket limiter, in-memory TTL cache | Lightweight deployments without external infra | Use `thin` under a router to prevent double-stacking resiliency. Use `standalone` for prototyping or single-service deployments. --- ## Testing & Conformance ### One-Command Testing ```bash # All protocols at once make test-all-conformance # Specific protocols make test-llm-conformance make test-vector-conformance make test-graph-conformance make test-embedding-conformance ``` ### CLI ```bash corpus-sdk verify # All protocols corpus-sdk verify -p llm -p vector # Selected protocols corpus-sdk test-llm-conformance # Single protocol ``` ### Direct pytest ```bash pytest tests/ -v --cov=corpus_sdk --cov-report=html ``` Requirements for "CORPUS-Compatible" certification are in [`docs/conformance/CERTIFICATION.md`](docs/conformance/CERTIFICATION.md). --- ## 📚 Documentation Layout **Spec (normative):** [`docs/spec/`](docs/spec/) | File | Contents | |---|---| | `SPECIFICATION.md` | Full protocol suite specification (all domains, cross-cutting behavior) | | `PROTOCOL.md` | Wire-level envelopes, streaming semantics, canonical `op` registry | | `ERRORS.md` | Canonical error taxonomy & mapping rules | | `METRICS.md` | Metrics schema & SIEM-safe observability | | `SCHEMA.md` | JSON/type shapes | | `VERSIONING.md` | Semantic versioning & compatibility rules | **Guides (how-to):** [`docs/guides/`](docs/guides/) | File | Contents | |---|---| | `QUICK_START.md` | End-to-end flows for all four protocols | | `IMPLEMENTATION.md` | How to implement adapters | | `ADAPTER_RECIPES.md` | Real-world multi-cloud and RAG scenarios | | `CONFORMANCE_GUIDE.md` | How to run & interpret conformance suites | **Conformance (testing):** [`docs/conformance/`](docs/conformance/) — Per-protocol test specs, schema conformance, behavioral conformance, and certification requirements. --- ## FAQ <details> <summary><strong>Is the SDK open source?</strong></summary> Yes. The SDK (protocols, bases, example adapters) is open source under Apache-2.0. CORPUS Router and official production adapters are commercial. </details> <details> <summary><strong>Do I have to use CORPUS Router?</strong></summary> No. The SDK composes with any router or control plane. CORPUS Router is optional and adheres to the same public protocols. </details> <details> <summary><strong>How does CORPUS relate to LangChain / LlamaIndex / MCP / OpenRouter?</strong></summary> They're complementary. LangChain/LlamaIndex are application frameworks. MCP standardizes tools and data sources. OpenRouter unifies LLM providers. CORPUS standardizes the infrastructure layer (LLM + Vector + Graph + Embedding) underneath all of them with consistent errors, metrics, and capabilities discovery. </details> <details> <summary><strong>Why async-only?</strong></summary> Modern AI workloads require high concurrency. Async-first prevents blocking the event loop. Sync wrappers can be built on top if needed. </details> <details> <summary><strong>What happens if my adapter raises a non-normalized error?</strong></summary> Bases catch unexpected exceptions and record them as `UnhandledException` in metrics. Wrap provider errors in normalized exceptions for proper handling. </details> <details> <summary><strong>Can CORPUS Router run on-prem?</strong></summary> Yes. Available as managed cloud or on-prem deployment for regulated and air-gapped environments. </details> --- ## Troubleshooting | Problem | Solution | |---|---| | Double-stacked resiliency (timeouts firing twice) | Ensure `mode="thin"` under your router | | Circuit breaker opens frequently | Reduce concurrency or switch to `thin` with external circuit breaker | | Cache returns stale results | Verify sampling params in cache key; check `cache_ttl_s` | | `DeadlineExceeded` on fast operations | Ensure `deadline_ms` is absolute epoch time, not relative. Check NTP sync. | | Health check failures | Inspect `_do_health` implementation; verify backend reachability and credentials | ```python # Debug mode import logging logging.basicConfig(level=logging.DEBUG) logging.getLogger("corpus_sdk").setLevel(logging.DEBUG) ``` --- ## Contributing ```bash git clone https://github.com/Corpus-OS/corpusos.git cd corpus-sdk pip install -e ".[dev]" pytest ``` Follow PEP-8 (ruff/black). Type hints required on all public APIs. Include tests for new features. Maintain low-cardinality metrics — never add PII to `extra` fields. Observe SemVer. We especially welcome **community adapter contributions** for new LLM, vector, graph, and embedding backends. Community questions: [GitHub Discussions](https://github.com/corpus/corpus-sdk/discussions) preferred. --- ## License & Commercial Options **License:** Apache-2.0 ([`LICENSE`](LICENSE)) | Need | Solution | Cost | |---|---|---| | Learning / prototyping | `corpus_sdk` + example adapters | **Free (OSS)** | | Production with your own infra | `corpus_sdk` + your adapters | **Free (OSS)** | | Production with official adapters | `corpus_sdk` + Official Adapters | **Commercial** | | Enterprise multi-provider | `corpus_sdk` + CORPUS Router (Managed or On-Prem) | **Commercial** | --- **Contact:** [team@corpusos.com](mailto:team@corpusos.com) **Website:** [https://corpusos.com](https://corpusos.com) **Docs:** [https://docs.corpusos.com](https://docs.corpusos.com) --- **Built by the Corpus OS team** — wire-level AI infrastructure you integrate once and connect anywhere.