Canopy (Pinecone RAG framework)

Canopy is Pinecone's open-source framework that bundles chunking, embedding, retrieval, and chat-history management into a single OpenAI-compatible chat server. If Pinecone is already chosen and you want a working RAG chat endpoint in under an hour, Canopy is the fastest path.

When to Use Canopy vs Custom

Canopy is maintenance-mode as of late 2024 — suitable for stable workloads but don't expect rapid new features. For a Pinecone-native stack needing active innovation, combine Pinecone directly with LangGraph/LlamaIndex.

Installation

pip install canopy-sdk

export PINECONE_API_KEY=...
export OPENAI_API_KEY=...
export ANTHROPIC_API_KEY=...   # optional, if using Anthropic LLM
export INDEX_NAME=my-knowledge-base

Create an Index

canopy new        # creates a Pinecone serverless index sized for text-embedding-3-small

Programmatic:

from canopy.knowledge_base import KnowledgeBase
from canopy.tokenizer import Tokenizer

Tokenizer.initialize()
kb = KnowledgeBase(index_name="my-knowledge-base")
kb.create_canopy_index()

Ingest Documents

CLI (one-shot):

canopy upsert ./data         # walks a directory; supports .jsonl / .parquet / .txt / .md

JSONL input format (one document per line):

{"id":"doc-1","text":"Canopy bundles chunking ...","source":"docs","metadata":{"team":"ml"}}

Programmatic upsert:

from canopy.models.data_models import Document

docs = [
    Document(
        id="doc-1",
        text="Canopy is Pinecone's RAG framework...",
        source="https://docs.pinecone.io/canopy",
        metadata={"team": "ml", "version": "v1"},
    ),
]
kb.upsert(docs, batch_size=100)

Upserts are idempotent by id; reuse stable IDs so re-ingests don't duplicate.

Query (Retrieval Only)

from canopy.models.data_models import Query

results = kb.query([Query(text="How do I configure chunking?", top_k=5)])
for r in results[0].documents:
    print(r.score, r.text[:120])

Chat (Retrieval + Generation)

from canopy.context_engine import ContextEngine
from canopy.chat_engine import ChatEngine
from canopy.models.data_models import UserMessage

context_engine = ContextEngine(kb)
chat_engine = ChatEngine(context_engine=context_engine)

history = [UserMessage(content="How does Canopy chunk Markdown files?")]
resp = chat_engine.chat(messages=history, stream=False)
print(resp.choices[0].message.content)

Token budget: Canopy automatically sizes the retrieval budget based on the model context window and reserves space for chat history.

Run the Server (OpenAI-compatible)

canopy start --port 8000
# OpenAI-compatible endpoints:
#   POST /v1/chat/completions
#   POST /context/query   (retrieval only)
#   POST /context/upsert

Point any OpenAI client at it:

from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="canopy")
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "What is Canopy?"}],
)

Because it's OpenAI-compatible, tools like Continue, LlamaIndex OpenAILike, LangChain ChatOpenAI, and LibreChat all work unchanged — just swap base_url.

Configuration (config.yaml)

tokenizer:
  type: OpenAITokenizer
  params:
    model_name: gpt-4o

chat_engine:
  max_prompt_tokens: 4096
  max_generated_tokens: 1024
  llm:
    type: AnthropicLLM
    params:
      model_name: claude-sonnet-4-5

  context_engine:
    knowledge_base:
      params:
        default_top_k: 10
      chunker:
        type: MarkdownChunker
        params:
          chunk_size: 512
          chunk_overlap: 50
      record_encoder:
        type: OpenAIRecordEncoder
        params:
          model_name: text-embedding-3-small

Launch with config:

canopy start --config config.yaml

Customization Hooks

Custom Chunker

from canopy.knowledge_base.chunker import Chunker
from canopy.models.data_models import Document, KBDocChunk

class HeaderAwareChunker(Chunker):
    def chunk_single_document(self, doc: Document) -> list[KBDocChunk]:
        # Split on H2 headers, emit one chunk per section
        chunks = []
        for i, section in enumerate(doc.text.split("\n## ")):
            chunks.append(KBDocChunk(
                id=f"{doc.id}_{i}",
                text=section,
                source=doc.source,
                document_id=doc.id,
                metadata=doc.metadata,
            ))
        return chunks

Register in config.yaml via chunker.type: HeaderAwareChunker (after importing the module).

Custom Record Encoder (embedding model)

from canopy.knowledge_base.record_encoder import DenseRecordEncoder
from sentence_transformers import SentenceTransformer

class BGERecordEncoder(DenseRecordEncoder):
    def __init__(self, model_name="BAAI/bge-large-en-v1.5", batch_size=32):
        super().__init__(dimension=1024, batch_size=batch_size)
        self._model = SentenceTransformer(model_name)
    def _encode_documents_batch(self, texts): return self._model.encode(texts).tolist()
    def _encode_queries_batch(self, texts):   return self._model.encode(texts).tolist()

Custom Context Builder

Override ContextEngine to inject reranking or hybrid search before returning the prompt context.

Chat History Management

Canopy trims older turns to fit the token budget automatically. For stable personas, pre-seed a system message; it is never trimmed.

from canopy.models.data_models import SystemMessage
history = [
    SystemMessage(content="You are a concise platform engineer."),
    UserMessage(content="Explain our release process."),
]

Anti-Patterns

Production Checklist

• Pinecone serverless or pod index sized for expected vector count
• Stable document ids so upserts are idempotent
• Per-tenant namespaces for isolation
• Config-as-code (config.yaml) versioned in git
• Auth/reverse proxy in front of canopy start
• Embedding model pinned in config.yaml + manifest
• Monitoring on Pinecone query p95 and error rate
• Migration plan if the project stays in maintenance mode