feat: route pagepiper.rag_query through cf-orch task assignment layer (closes #7)

When CF_ORCH_URL is set, chat now calls CFOrchClient.task_allocate("pagepiper",
"rag_query") instead of routing through LLMRouter with an explicit model. The
coordinator resolves the assignment (granite-4.1-8b via assignments.yaml) and
returns an allocated URL; pagepiper wraps it in a minimal LLMRouter config for
the Synthesizer. Falls back to LLMRouter on TaskNotFound or allocation failure,
so standalone Ollama installs are unaffected.

Extracts _run_chat() and _build_llm_for_alloc() helpers to keep the endpoint
body readable regardless of which path fires.
This commit is contained in:
pyr0ball 2026-05-16 21:05:02 -07:00
parent 9d33b1ab54
commit b16620385a

View file

@ -47,7 +47,7 @@ class ChatFeedbackRequest(BaseModel):
def _get_llm_router(): def _get_llm_router():
"""Return LLMRouter if Ollama configured, else None.""" """Return LLMRouter if Ollama/cf-orch configured, else None."""
from app.config import get_llm_config from app.config import get_llm_config
cfg = get_llm_config() cfg = get_llm_config()
@ -58,6 +58,65 @@ def _get_llm_router():
return LLMRouter(cfg) return LLMRouter(cfg)
def _build_llm_for_alloc(alloc) -> "LLMRouter":
"""Wrap a cf-orch task allocation in a minimal LLMRouter for completion calls."""
from circuitforge_core.llm import LLMRouter
base_url = alloc.url.rstrip("/")
if not base_url.endswith("/v1"):
base_url += "/v1"
cfg = {
"fallback_order": ["orch_task"],
"backends": {
"orch_task": {
"type": "openai_compat",
"base_url": base_url,
"model": alloc.model or "default",
"supports_images": False,
}
},
}
return LLMRouter(cfg)
def _run_chat(req: "ChatRequest", ctx: "UserCtx", llm) -> "ChatResponse":
retriever = Retriever(ctx.bm25)
chunks = retriever.hybrid_search(
query=req.message,
top_k=req.top_k,
doc_ids=req.doc_ids,
db_path=ctx.db_path,
vec_db_path=ctx.vec_db_path,
llm=llm,
)
if not chunks:
return ChatResponse(
answer=(
"I couldn't find any relevant passages. "
"Try a different query or check which documents are indexed."
),
citations=[],
)
synth = Synthesizer(llm)
result = synth.synthesize(
message=req.message,
history=[t.model_dump() for t in req.history],
chunks=chunks,
)
return ChatResponse(
answer=result.answer,
citations=[
{
"doc_id": c.doc_id,
"page_number": c.page_number,
"snippet": c.snippet,
"bm25_score": c.bm25_score,
}
for c in result.citations
],
)
def _require_llm(): def _require_llm():
"""Return LLMRouter or raise 402.""" """Return LLMRouter or raise 402."""
llm = _get_llm_router() llm = _get_llm_router()
@ -81,46 +140,24 @@ def chat(
ctx: UserCtx = Depends(get_user_ctx), ctx: UserCtx = Depends(get_user_ctx),
_tier: str = Depends(require_paid_tier), _tier: str = Depends(require_paid_tier),
) -> ChatResponse: ) -> ChatResponse:
orch_url = os.environ.get("CF_ORCH_URL", "").strip()
if orch_url:
try:
from circuitforge_orch.client import CFOrchClient, TaskNotFound # type: ignore[import]
api_key = os.environ.get("CF_LICENSE_KEY", "")
client = CFOrchClient(orch_url, api_key=api_key)
with client.task_allocate("pagepiper", "rag_query") as alloc:
llm = _build_llm_for_alloc(alloc)
return _run_chat(req, ctx, llm)
except Exception as exc:
logger.warning(
"cf-orch task allocation for pagepiper.rag_query failed, falling back to LLMRouter: %s",
exc,
)
llm = _require_llm() llm = _require_llm()
return _run_chat(req, ctx, llm)
retriever = Retriever(ctx.bm25)
chunks = retriever.hybrid_search(
query=req.message,
top_k=req.top_k,
doc_ids=req.doc_ids,
db_path=ctx.db_path,
vec_db_path=ctx.vec_db_path,
llm=llm,
)
if not chunks:
return ChatResponse(
answer=(
"I couldn't find any relevant passages. "
"Try a different query or check which documents are indexed."
),
citations=[],
)
synth = Synthesizer(llm)
result = synth.synthesize(
message=req.message,
history=[t.model_dump() for t in req.history],
chunks=chunks,
)
return ChatResponse(
answer=result.answer,
citations=[
{
"doc_id": c.doc_id,
"page_number": c.page_number,
"snippet": c.snippet,
"bm25_score": c.bm25_score,
}
for c in result.citations
],
)
@router.get("/feedback/status") @router.get("/feedback/status")