---

name: canopy description: | Pinecone's Canopy framework — a pre-packaged RAG stack on top of Pinecone. Covers Canopy CLI, REST server (OpenAI-compatible), chunking, embedding, retrieval, chat-history management, and customization hooks (custom chunkers/records).

USE WHEN: user mentions "Canopy", "Pinecone Canopy", "canopy start", "canopy new", "Pinecone out-of-the-box RAG"

DO NOT USE FOR: custom pipelines on non-Pinecone stores - use rag-architecture; Postgres-based stacks - use r2r; LangChain/LlamaIndex abstractions - use langgraph-rag/llamaindex allowed-tools: Read, Grep, Glob, Write, Edit

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

Signal	Canopy	Custom (LangChain / LlamaIndex)
Vector store is Pinecone	Strong fit	Either works
Need an OpenAI-compatible chat endpoint today	Yes	Custom work
Mostly defaults work	Yes	Custom
Need graph RAG, advanced reranking, custom agents	No	Yes
Non-Pinecone vector DB	No	Yes
Want to version the whole RAG stack as one service	Yes	Manual

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

Anti-Pattern	Fix
Hardcoding index name in client code	Read from INDEX_NAME env
Running canopy start without auth	Put behind reverse proxy with auth (OIDC/API key)
Using default chunker for code/tables	Swap in MarkdownChunker or custom CodeChunker
One Pinecone index for all tenants without namespaces	Use namespace=tenant_id on upsert/query
Ignoring the maintenance-mode status	Plan a migration path if the project stalls
Re-embedding with a different model in-place	Treat as a reindex — new index, dual-write, cutover

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