Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -102,8 +102,8 @@ flowchart TB

subgraph RETRIEVE["⑤ Retrieval (shared)"]
Query["POST /v1|/v2 retrieval/query"] --> Pipeline["run_retrieval_query"]
Pipeline --> Classic["classic_topk / small_corpus (use_agentic=False)"]
Pipeline --> Explore["agent_explore (use_agentic unset/true)"]
Pipeline --> Classic["classic_topk / small_corpus"]
Pipeline --> Explore["agent_explore"]
Explore --> Cursor["AGENT_EXPLORE_HARNESS=cursor_sdk"]
Explore --> Own["AGENT_EXPLORE_HARNESS=openai"]
Classic --> Channels["map_unit_discovery: path+content BM25 -> RRF"]
Expand Down Expand Up @@ -542,7 +542,7 @@ Core retrieval internals are grouped by ownership:

### Retrieval Switch and Agent Modes

Per-request `use_agentic` only turns agent retrieval on or off: `False` → classic top-K (map-unit BM25); `None`/`True` → agent_explore. When agent retrieval is on, `AGENT_EXPLORE_HARNESS` selects Cursor (`cursor_sdk`) or Knowhere's own harness (`openai`).
Per-request `use_agentic`: `False` → classic top-K (map-unit BM25); `True` → agent_explore; unset (`None`) → a query router chooses classic or agent. When agent retrieval is on, `AGENT_EXPLORE_HARNESS` selects Cursor (`cursor_sdk`) or Knowhere's own harness (`openai`).

#### Classic Mode (map-unit BM25)

Expand Down
5 changes: 4 additions & 1 deletion apps/api/.env.example
Original file line number Diff line number Diff line change
Expand Up @@ -103,10 +103,13 @@ ARK_API_KEY=
# answer_text is always empty.
# use_agentic on the request turns agent retrieval on or off:
# false → no agent; classic map-unit BM25
# omit/true → agent on; AGENT_EXPLORE_HARNESS selects the mode
# true → agent on; AGENT_EXPLORE_HARNESS selects the mode
# omit → query router chooses classic or agent
# cursor_sdk → Cursor (needs CURSOR_API_KEY)
# openai → Knowhere's own harness (DS_KEY / ALI_API_KEYS)
# AGENT_EXPLORE_HARNESS=cursor_sdk
# RETRIEVAL_QUERY_ROUTER_ENABLED=true
# RETRIEVAL_QUERY_ROUTER_MODEL=
# CURSOR_API_KEY= # only when harness is cursor_sdk
# AGENT_EXPLORE_CURSOR_MODEL=composer-2.5
# Classic BM25 may use Postgres FTS prefilter:
Expand Down
13 changes: 11 additions & 2 deletions apps/api/app/api/v1/routes/retrieval.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,8 +73,9 @@ class RetrievalQueryRequest(BaseModel):
None,
description=(
"Turns agent retrieval on or off. False uses classic map-unit "
"BM25. Unset or true turns agent retrieval on; "
"AGENT_EXPLORE_HARNESS then selects Cursor (cursor_sdk) or "
"BM25. True turns agent retrieval on. Unset (None) lets the "
"query router choose classic or agent. When agent retrieval is "
"on, AGENT_EXPLORE_HARNESS then selects Cursor (cursor_sdk) or "
"Knowhere's own harness (openai)."
),
)
Expand Down Expand Up @@ -156,6 +157,14 @@ class RetrievalQueryResponse(BaseModel):
"observation/decision/result schema."
),
)
router_decision: dict[str, Any] | None = Field(
default=None,
description=(
"Present when use_agentic was unset and the query router chose "
"agent or classic. Fields: mode, reason, source, model_name, "
"tokens_used."
),
)


async def execute_retrieval_query(
Expand Down
299 changes: 299 additions & 0 deletions apps/api/tests/contract/test_retrieval_query_router_contract.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,299 @@
from __future__ import annotations

from collections.abc import Callable
from contextlib import AbstractAsyncContextManager
from typing import Any, cast
from uuid import uuid4

import pytest
from httpx import AsyncClient
from pytest import MonkeyPatch

from shared.services.retrieval.routing.types import QueryMode, RouterDecision
from tests.contract.test_retrieval_classic_map_unit_contract import _publish_document
from tests.contract.test_retrieval_contract import (
_episode_keeping_refs,
_patch_harness,
_seed_retrieval_document,
)


def _decision(*, mode: QueryMode, source: str = "llm") -> RouterDecision:
return RouterDecision(
mode=mode,
reason="contract",
source=source,
model_name="test-router-model",
tokens_used=2,
)


async def _publish_above_top_k(*, namespace: str) -> dict[str, str]:
identifier = uuid4().hex[:8]
chunk_id = f"hit-{identifier}"
target = await _publish_document(
namespace=namespace,
source_file_name="target.pdf",
chunks=[
{
"chunk_id": chunk_id,
"type": "text",
"content": "unique alpha ranking marker",
"path": "target.pdf/Root/Hit/body",
"order": 1,
"metadata": {},
}
],
)
await _publish_document(
namespace=namespace,
source_file_name="filler.pdf",
chunks=[
{
"chunk_id": f"filler-{identifier}",
"type": "text",
"content": "unrelated filler content",
"path": "filler.pdf/Root/Noise/body",
"order": 1,
"metadata": {},
}
],
)
return {
**target,
"chunk_id": chunk_id,
"section_path": "Root / Hit / body",
}


def _patch_classifier(
monkeypatch: MonkeyPatch,
decision: RouterDecision,
calls: list[str],
) -> None:
from shared.services.retrieval.execution import routes as retrieval_routes

async def fake_classify(query: str) -> RouterDecision:
calls.append(query)
return decision

monkeypatch.setattr(retrieval_routes, "classify_query_mode", fake_classify)


def _patch_classifier_boom(monkeypatch: MonkeyPatch, calls: list[str], message: str) -> None:
from shared.services.retrieval.execution import routes as retrieval_routes

async def boom(_query: str) -> RouterDecision:
calls.append("called")
raise AssertionError(message)

monkeypatch.setattr(retrieval_routes, "classify_query_mode", boom)


@pytest.mark.asyncio
async def test_unset_use_agentic_classic_decision_uses_classic_route(
developer_api_client_factory: Callable[
[], AbstractAsyncContextManager[AsyncClient]
],
monkeypatch: MonkeyPatch,
) -> None:
namespace = f"router-classic-{uuid4().hex[:8]}"
calls: list[str] = []

async with developer_api_client_factory() as api_client:
await _publish_above_top_k(namespace=namespace)
_patch_classifier(monkeypatch, _decision(mode=QueryMode.CLASSIC), calls)
response = await api_client.post(
"/api/v1/retrieval/query",
json={
"namespace": namespace,
"query": "unique alpha ranking",
"top_k": 1,
},
)

assert response.status_code == 200
body = cast(dict[str, Any], response.json())
assert calls == ["unique alpha ranking"]
assert body["router_used"] == "classic_topk"
assert body["router_decision"] == {
"mode": "classic",
"reason": "contract",
"source": "llm",
"model_name": "test-router-model",
"tokens_used": 2,
}


@pytest.mark.asyncio
async def test_unset_use_agentic_agent_decision_uses_agent_explore(
developer_api_client_factory: Callable[
[], AbstractAsyncContextManager[AsyncClient]
],
monkeypatch: MonkeyPatch,
) -> None:
namespace = f"router-agent-{uuid4().hex[:8]}"
calls: list[str] = []

async with developer_api_client_factory() as api_client:
target = await _publish_above_top_k(namespace=namespace)
_patch_classifier(monkeypatch, _decision(mode=QueryMode.AGENT), calls)
_patch_harness(
monkeypatch,
_episode_keeping_refs(
documents=[target],
notes="unique alpha ranking marker",
),
)
response = await api_client.post(
"/api/v1/retrieval/query",
json={
"namespace": namespace,
"query": "unique alpha ranking",
"top_k": 1,
},
)

assert response.status_code == 200
body = cast(dict[str, Any], response.json())
assert calls == ["unique alpha ranking"]
assert body["router_used"] == "agent_explore"
assert body["router_decision"]["mode"] == "agent"
assert body["router_decision"]["source"] == "llm"


@pytest.mark.asyncio
async def test_explicit_use_agentic_does_not_call_classifier(
developer_api_client_factory: Callable[
[], AbstractAsyncContextManager[AsyncClient]
],
monkeypatch: MonkeyPatch,
) -> None:
namespace = f"router-explicit-{uuid4().hex[:8]}"
calls: list[str] = []

async with developer_api_client_factory() as api_client:
target = await _publish_above_top_k(namespace=namespace)
_patch_classifier_boom(
monkeypatch,
calls,
"classifier must not run when use_agentic is set",
)
classic = await api_client.post(
"/api/v1/retrieval/query",
json={
"namespace": namespace,
"query": "unique alpha ranking",
"top_k": 1,
"use_agentic": False,
},
)
_patch_harness(
monkeypatch,
_episode_keeping_refs(documents=[target], notes="explicit agent"),
)
agent = await api_client.post(
"/api/v1/retrieval/query",
json={
"namespace": namespace,
"query": "unique alpha ranking",
"top_k": 1,
"use_agentic": True,
},
)

assert classic.status_code == 200
assert agent.status_code == 200
classic_body = cast(dict[str, Any], classic.json())
agent_body = cast(dict[str, Any], agent.json())
assert calls == []
assert classic_body["router_used"] == "classic_topk"
assert classic_body.get("router_decision") is None
assert agent_body["router_used"] == "agent_explore"
assert agent_body.get("router_decision") is None


@pytest.mark.asyncio
async def test_disabled_router_keeps_unset_use_agentic_on_agent(
developer_api_client_factory: Callable[
[], AbstractAsyncContextManager[AsyncClient]
],
monkeypatch: MonkeyPatch,
) -> None:
namespace = f"router-disabled-{uuid4().hex[:8]}"
calls: list[str] = []

async with developer_api_client_factory() as api_client:
target = await _publish_above_top_k(namespace=namespace)
monkeypatch.setattr(
"shared.services.retrieval.routing.config.QUERY_ROUTER_ENABLED",
False,
)
_patch_classifier_boom(
monkeypatch,
calls,
"classifier must not run when the router is disabled",
)
_patch_harness(
monkeypatch,
_episode_keeping_refs(documents=[target], notes="disabled router"),
)
response = await api_client.post(
"/api/v1/retrieval/query",
json={
"namespace": namespace,
"query": "unique alpha ranking",
"top_k": 1,
},
)

assert response.status_code == 200
body = cast(dict[str, Any], response.json())
assert calls == []
assert body["router_used"] == "agent_explore"
assert body["router_decision"] == {
"mode": "agent",
"reason": "",
"source": "disabled",
"model_name": "",
"tokens_used": 0,
}


@pytest.mark.asyncio
async def test_small_corpus_skips_query_router(
developer_api_client_factory: Callable[
[], AbstractAsyncContextManager[AsyncClient]
],
monkeypatch: MonkeyPatch,
) -> None:
namespace = f"router-small-{uuid4().hex[:8]}"
calls: list[str] = []

async with developer_api_client_factory() as api_client:
await _seed_retrieval_document(
user_id="local-dev-user",
namespace=namespace,
source_file_name="only.pdf",
section_path="only/section",
content="only chunk",
)
_patch_classifier_boom(
monkeypatch,
calls,
"classifier must not run on the small-corpus path",
)
response = await api_client.post(
"/api/v1/retrieval/query",
json={
"namespace": namespace,
"query": "only chunk",
"top_k": 10,
},
)

assert response.status_code == 200
body = cast(dict[str, Any], response.json())
assert calls == []
assert body["router_used"] == "small_corpus_all"
assert body.get("router_decision") is None
5 changes: 4 additions & 1 deletion apps/worker/.env.example
Original file line number Diff line number Diff line change
Expand Up @@ -97,10 +97,13 @@ ARK_API_KEY=
# answer_text is always empty.
# use_agentic on the request turns agent retrieval on or off:
# false → no agent; classic map-unit BM25
# omit/true → agent on; AGENT_EXPLORE_HARNESS selects the mode
# true → agent on; AGENT_EXPLORE_HARNESS selects the mode
# omit → query router chooses classic or agent
# cursor_sdk → Cursor (needs CURSOR_API_KEY)
# openai → Knowhere's own harness (DS_KEY / ALI_API_KEYS)
# AGENT_EXPLORE_HARNESS=cursor_sdk
# RETRIEVAL_QUERY_ROUTER_ENABLED=true
# RETRIEVAL_QUERY_ROUTER_MODEL=
# CURSOR_API_KEY= # only when harness is cursor_sdk
# AGENT_EXPLORE_CURSOR_MODEL=composer-2.5
# Classic BM25 may use Postgres FTS prefilter:
Expand Down
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
"""In-process tool-calling agentic retrieval route.

Default agentic path when ``use_agentic`` is unset or true. Runs the
``agent_tools`` corpus registry through an LLM tool-calling loop. Map-nav is
archived under ``deprecated/mapnav/`` and is not a live route.
Agentic path when ``use_agentic`` is true, or when it is unset and the query
router selects agent. Runs the ``agent_tools`` corpus registry through an LLM
tool-calling loop. Map-nav is archived under ``deprecated/mapnav/`` and is
not a live route.

Which provider runs that loop (Cursor SDK or OpenAI-compatible) is the
``AGENT_EXPLORE_HARNESS`` switch resolved via ``resolve_harness()``.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ async def project_public_retrieval_response(response: dict[str, Any]) -> dict[st
public_response['failure_reason'] = response['failure_reason']
if response.get('decision_trace') is not None:
public_response['decision_trace'] = response['decision_trace']
if response.get('router_decision') is not None:
public_response['router_decision'] = response['router_decision']

projected_rows = await enrich_rows_with_retrieval_asset_url(
response.get('results', []),
Expand Down
Loading
Loading