Quickstart

pip install pinecone anthropic

import anthropic
from pinecone import Pinecone

pc = Pinecone(api_key="{{YOUR_API_KEY}}")
llm = anthropic.Anthropic(api_key="YOUR_ANTHROPIC_API_KEY")

# Create an index with integrated embedding
if not pc.has_index("knowledge"):
    pc.create_index_for_model(
        name="knowledge",
        cloud="aws",
        region="us-east-1",
        embed={
            "model": "llama-text-embed-v2",
            "field_map": {"text": "content"}
        }
    )

index = pc.Index("knowledge")

# Load your knowledge base
index.upsert_records(
    namespace="docs",
    records=[
        {"_id": "policy-1", "content": "Refund requests must be submitted within 30 days of purchase. After 30 days, only store credit is available.", "category": "policies"},
        {"_id": "policy-2", "content": "Enterprise customers get dedicated support with a 4-hour response time SLA. Standard support responds within 24 hours.", "category": "policies"},
        {"_id": "spec-1", "content": "The WonderVector 5000 supports up to 100,000 vectors per namespace with a maximum dimensionality of 4096.", "category": "specs"},
        {"_id": "spec-2", "content": "API rate limits: Free tier is 100 requests/minute, Pro tier is 1000 requests/minute, Enterprise is unlimited with fair use.", "category": "specs"},
        {"_id": "spec-3", "content": "Data is encrypted at rest using AES-256 and in transit using TLS 1.3. SOC2 Type II compliance is maintained.", "category": "security"},
        {"_id": "hr-1", "content": "New employees receive 15 days PTO in their first year, increasing to 20 days after 2 years and 25 days after 5 years.", "category": "hr"},
        {"_id": "hr-2", "content": "The company matches 401k contributions up to 4% of salary. Vesting is immediate for all employees.", "category": "hr"},
        {"_id": "proc-1", "content": "To request a new software license, submit a ticket in the IT portal. Approvals take 2-3 business days for standard software.", "category": "procedures"},
        {"_id": "proc-2", "content": "Production deployments require approval from the team lead and a passing CI/CD pipeline. Hotfixes can bypass the lead approval.", "category": "procedures"},
        {"_id": "proc-3", "content": "Vendor invoices over $10,000 require VP approval. Under $10,000 requires manager approval only.", "category": "procedures"},
    ]
)

tools = [{
    "name": "search_knowledge_base",
    "description": "Search the company knowledge base for policies, specs, HR info, and procedures.",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "The search query"}
        },
        "required": ["query"]
    }
}]

def ask(question: str) -> str:
    messages = [{"role": "user", "content": question}]

    # disable_parallel_tool_use keeps this loop simple: with parallel calls,
    # every tool_use block would need a matching tool_result in the next message
    response = llm.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        tools=tools,
        tool_choice={"type": "auto", "disable_parallel_tool_use": True},
        messages=messages
    )

    # If the model wants to use a tool, call it and return the result
    while response.stop_reason == "tool_use":
        tool_block = next(b for b in response.content if b.type == "tool_use")
        tool_result = search_knowledge_base(tool_block.input["query"])

        messages += [
            {"role": "assistant", "content": response.content},
            {"role": "user", "content": [{
                "type": "tool_result",
                "tool_use_id": tool_block.id,
                "content": tool_result
            }]}
        ]

        response = llm.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            tools=tools,
            tool_choice={"type": "auto", "disable_parallel_tool_use": True},
            messages=messages
        )

    return next(b.text for b in response.content if hasattr(b, "text"))