> ## Documentation Index
> Fetch the complete documentation index at: https://docs.hydradb.com/llms.txt
> Use this file to discover all available pages before exploring further.

> ## Agent Instructions
> AGENTS.md is HydraDB's maintained agent guide. Read https://docs.hydradb.com/AGENTS.md once, then use its content. /llms.txt and /llms-full.txt redirect to the same guide; they are not separate documentation indexes. If the guide is already loaded, continue with it instead of fetching either alias or reloading the guide.

# Build your own competitive intelligence agent

> A tracker that ingests competitor pricing pages, releases, and job posts, and always answers from the current version.

Competitor signals go stale fast: a pricing page changes, and last quarter's version is now the wrong answer that looks most like the right one. This cookbook builds a tracker around two ideas: a stable ID per page, so a new version replaces the old one, and a metadata filter per competitor, so one company's signals never answer a question about another.

| Benchmark | Result | What it shows here |
| - | - | - |
| [KyroBench](https://benchmarks.hydradb.com/kyrobench) | 0.776 nDCG, 0.338 above the best baseline | The current, in-scope source ranked above stale and out-of-scope look-alikes |
| [FinanceBench](https://benchmarks.hydradb.com/financebench) | 91.4% recall\@10 | Exact figures found in dense filings |

## Prerequisites

* A HydraDB API key from [dashboard.hydradb.com](https://dashboard.hydradb.com), set as `HYDRADB_API_KEY`.
* An OpenAI API key, set as `OPENAI_API_KEY`. HydraDB returns context, not answers, so any model works; swap the `answer` function below for your own.
* Python 3.10 or later, with the SDKs: `pip install "hydradb-sdk>=2.1.7,<3" openai`.

Save this as `hydra.py`. Every cookbook imports it. It holds what every HydraDB application does around its own logic: wait for the database, check what an ingest accepted, wait for indexing, and pass query results to a model.

```python hydra.py theme={"dark"}
import os
import time

from hydra_db import HydraDB
from hydra_db.errors import ConflictError
from hydra_db.helpers import build_string
from openai import OpenAI

client = HydraDB(token=os.environ["HYDRADB_API_KEY"])
llm = OpenAI()


def create_database(database, schema=None):
    """Create the database if it is new, then wait until it accepts context."""
    try:
        if schema:
            client.databases.create(database=database, database_metadata_schema=schema)
        else:
            client.databases.create(database=database)
    except ConflictError:
        pass  # The database already exists.
    while not client.databases.status(database=database).data.infra.ready_for_ingestion:
        time.sleep(5)


def source_ids(result):
    """Return the IDs an ingest accepted. Raise if it rejected any item."""
    for item in result.data.results:
        if item.error_code:
            raise RuntimeError(f"{item.filename or item.id}: {item.error_code} {item.error}")
    return [item.id for item in result.data.results]


def wait_indexed(database, collection, ids):
    """Block until every source has finished indexing. Indexing runs in the background."""
    while True:
        statuses = client.context.status(
            database=database, collection=collection, ids=ids
        ).data.statuses
        for status in statuses:
            if status.indexing_status == "errored":
                raise RuntimeError(f"{status.error_code}: {status.error_message}")
        if all(s.indexing_status == "completed" for s in statuses):
            return
        time.sleep(2)


def answer(question, result, instructions=""):
    """Send a query result to the model and return its answer."""
    completion = llm.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "Answer using only the context provided. If the answer "
                "is not in the context, say you don't know. " + instructions,
            },
            {
                "role": "user",
                "content": f"Context:\n{build_string(result)}\n\nQuestion: {question}",
            },
        ],
    )
    return completion.choices[0].message.content
```

## 1. Create a database with a metadata schema

Declare the fields you will filter on when you create the database. [Metadata filters](https://docs.hydradb.com/essentials/v2/metadata.md) work only on declared fields.

```python intel.py theme={"dark"}
import json

from hydra import answer, client, create_database, source_ids, wait_indexed

DATABASE, COLLECTION = "market_intel", "signals"


def text_field(name):
    return {
        "name": name,
        "data_type": "VARCHAR",
        "max_length": 128,
        "enable_dense_embedding": False,
        "enable_sparse_embedding": False,
    }


create_database(DATABASE, schema=[text_field("competitor"), text_field("signal")])
```

## 2. Ingest signals with stable IDs

Send each page as an [app source](https://docs.hydradb.com/essentials/v2/context.md#send-app-sources) with the competitor and signal type in `metadata`. Build the `id` from the page, not from the day you fetched it: ingesting the same `id` again replaces the stored version, because `upsert` defaults to `true`.

```python intel.py theme={"dark"}
def signal(source_id, competitor, kind_of_signal, title, body, fetched_at, url):
    return {
        "id": source_id,
        "title": title,
        "kind": "knowledge_base",
        "provider": "web",
        "external_id": source_id,
        "url": url,
        "timestamp": fetched_at,
        "fields": {"kind": "knowledge_base", "title": title, "body": body, "updated_at": fetched_at},
        "metadata": {"competitor": competitor, "signal": kind_of_signal},
    }


def ingest(signals):
    result = client.context.ingest(
        type="knowledge",
        database=DATABASE,
        collection=COLLECTION,
        app_knowledge=json.dumps(signals),
    )
    # A source must finish indexing before the same id is ingested again.
    wait_indexed(DATABASE, COLLECTION, source_ids(result))


ingest([
    signal("northwind_pricing", "northwind", "pricing", "Northwind pricing",
           "Team plan: $20 per seat per month. Annual billing only.",
           "2026-07-01T00:00:00Z", "https://northwind.example/pricing"),
    signal("northwind_job_ml_lead", "northwind", "hiring", "Northwind: ML Lead, Search",
           "Lead a new team building semantic search for the Northwind platform.",
           "2026-09-20T00:00:00Z", "https://northwind.example/careers/ml-lead"),
    signal("contoso_pricing", "contoso", "pricing", "Contoso pricing",
           "Team plan: $18 per seat per month, billed monthly or annually.",
           "2026-08-15T00:00:00Z", "https://contoso.example/pricing"),
])

# The Northwind pricing page changed. Same id, so this replaces the July version.
ingest([
    signal("northwind_pricing", "northwind", "pricing", "Northwind pricing",
           "Team plan: $24 per seat per month. Monthly billing now available.",
           "2026-10-05T00:00:00Z", "https://northwind.example/pricing"),
])
```

## 3. Query one competitor at a time

Filter on `competitor` so the two near-identical pricing pages cannot be confused. These are the KyroBench settings: hybrid matching, `fast` mode, eight results, graph context off, and a metadata filter for scope. In that report, removing the scope filter dropped recall from 0.917 to 0.185.

```python intel.py theme={"dark"}
def ask(competitor, question):
    result = client.query(
        database=DATABASE,
        collection=COLLECTION,
        query=question,
        type="knowledge",
        query_by="hybrid",
        mode="fast",
        graph_context=False,
        max_results=8,
        metadata_filters={"competitor": {"equals": competitor}},
    )
    return answer(question, result, "Cite the source title and its date for every claim.")


print(ask("northwind", "What does the Team plan cost?"))
```

## 4. Write the weekly digest

For a question that spans every signal type, keep the `competitor` filter, switch to `thinking` mode, and raise `recency_bias` so newer signals rank first.

```python intel.py theme={"dark"}
def digest(competitor):
    question = f"What changed at {competitor} recently, and what does it suggest about their roadmap?"
    result = client.query(
        database=DATABASE,
        collection=COLLECTION,
        query=question,
        type="knowledge",
        mode="thinking",
        recency_bias=0.7,
        metadata_filters={"competitor": {"equals": competitor}},
    )
    return answer(question, result, "Separate what the sources state from what you infer.")


print(digest("northwind"))
```

## Run it

```bash theme={"dark"}
python intel.py
```

The pricing answer should give $24, not $20 and not Contoso's \$18. The digest should connect the price rise with the search hiring, and label the roadmap reading as inference.

## Use real data

Run your fetcher on a schedule and call `ingest` with whatever it finds. Unchanged pages can be sent again safely. To be told when a batch finishes indexing instead of polling, register a [webhook](https://docs.hydradb.com/api-reference/v2/endpoint/webhooks-overview.md).

## Common mistakes

* **Putting the date in the `id`:** every fetch then adds a new source, and the stale versions compete with the current one.
* **Filtering on an undeclared field:** the filter matches nothing. Declare the field in the metadata schema first.
* **A bare filter value:** write `{"competitor": {"equals": "northwind"}}`. Filters without an operator are deprecated.

## Next steps

* [Metadata Filters](https://docs.hydradb.com/essentials/v2/metadata.md): operators, multi-value fields, and adding schema fields later.
* [Query](https://docs.hydradb.com/essentials/v2/query.md): `recency_bias` and when to use each `mode`.
* [Build your own Glean](https://docs.hydradb.com/cookbooks/v2/glean.md): the same pattern for knowledge inside your company.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.