Quickstart

import os
from mixpeek import Mixpeek

client = Mixpeek(api_key=os.environ["MIXPEEK_API_KEY"])

# Managed namespace — `namespace_name` is the only required field. You don't
# register extractors up front; the collection's `features` (below) auto-provision
# whatever pipeline the namespace needs.
ns = client.namespaces.create(namespace_name="agent-video-demo")

# Scope the client to the namespace you just created — every call below sends it
# as the X-Namespace header, which bucket/collection create and search require.
client.namespace = ns.namespace_id

bucket = client.buckets.create(
    bucket_name="demo-videos",
    # "video" is the standard source property name per modality —
    # collections created with `features` wire to it automatically.
    bucket_schema={"properties": {"video": {"type": "video"}}},
)

# Create the collection by picking a feature — Mixpeek resolves the
# pipeline and default input wiring (see /processing/features).
col = client.collections.create(
    collection_name="video-scenes",
    source={"type": "bucket", "bucket_ids": [bucket.bucket_id]},
    features=["video_search"],
)

# Upload the video to the bucket, then trigger the collection to process it.
client.buckets.upload(
    bucket.bucket_id,
    url="https://storage.googleapis.com/mixpeek-public-demo/videos/sample-product-demo.mp4",
)
result = client.collections.trigger(col.collection_id)

from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

ret = client.retrievers.create(
    retriever_name="agent-video-search",
    input_schema={"query_text": {"type": "text", "required": True}},
    collection_identifiers=[col.collection_id],
    stages=[{
        "stage_name": "search",
        "stage_type": "filter",
        "config": {
            "stage_id": "feature_search",
            "parameters": {
                "searches": [{
                    "feature_uri": "mixpeek://universal_extractor@v1/gemini-embedding-2",
                    "query": {"input_mode": "text", "value": "{{INPUT.query_text}}"},
                    "top_k": 50
                }],
                "final_top_k": 20
            }
        }
    }]
)

def search_video(query: str) -> str:
    results = client.retrievers.execute(
        retriever_id=ret.retriever_id,
        inputs={"query_text": query},
    )
    return "\n".join(
        f"[{r.get('start_time_s','?')}s] (score: {r.score:.3f}) {r.get('description','')}"
        for r in results.documents
    ) or "No results found."

video_tool = Tool(name="video_search", description="Search indexed video by natural language", func=search_video)
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "You answer questions about video content. Use video_search to find relevant moments."),
    ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_openai_tools_agent(llm, [video_tool], prompt)
executor = AgentExecutor(agent=agent, tools=[video_tool], verbose=True)
print(executor.invoke({"input": "What product features are shown?"})["output"])