Phase 9's operator path, built ahead of Phase 8 at the customer's direction and live at api.yokogawa.tech/documents. Upload, convert, review, approve, withdraw and restore. The pool screen is explicitly out of scope. Served by ai-api rather than ai-web, and mounted at /documents rather than /docs. ai.yokogawa.tech is SCADA-only since 2026-08-28 and passes through no Authelia, so it has no identity to record; publishers arrive on api.yokogawa.tech where the forward-auth headers still do. /docs stays with Swagger, which the customer is keeping - two things under one prefix with two different access policies is what gets misread during a later edit. Conversion is text extraction, not document parsing: pypdf, python-docx and openpyxl. Docling would be better at this and pulls torch, which lin001 has neither the memory to install nor the business running next to the demo plant's PLC. The cost is real - no layout, no table structure, and a scan cannot be read at all, so it is refused rather than stored empty. It is acceptable only because the converted text is shown to a person before the document can be cited, which is the same safety net the design already required for the header. convert.py is the one file to change if that stops being true. Chunking is mirrored from ingest.py rather than shared, because the two live in different images. They must stay identical: if they drift, the same document chunks differently depending on who loaded it, and the assistant answers or fails to answer depending on that. The step-sequence rule is locked by a test. Identity is self-asserted for the demo - the actor is typed on the form, which section 16 forbids, and the publisher list is one name with no password. Rows are written as `demo:<name>` with actor_groups = 'DEMO-UNVERIFIED' so that when real auth goes on, a name somebody typed stays tellable from a name Authelia proved. doc_actions cannot be deleted from, so an ambiguity there would be permanent. Two rules the code enforces rather than documents: uploading is open to anyone who reaches the page, because uploading changes nothing an operator can see - approving does, and that is what is gated; and an empty publisher list means nobody, not everybody. Verified on the host end to end: withdraw as a non-publisher 403s, with a short reason 400s, and as admin flips 5 chunks and writes a complete audit row; restore puts them back and keeps both rows. The corpus is unchanged afterwards. Requirements are split so the document dependencies install in their own layer - a change there costs four small wheels instead of re-resolving fastapi, langgraph and langfuse on a 2 vCPU shared host. The five divergences from section 16 are recorded in section 14. The one with teeth: files published through the UI stay in the inbox, so `ai-ingest --all` cannot see them and the two paths must not be used on the same document. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
210 lines
7.6 KiB
Python
210 lines
7.6 KiB
Python
"""FastAPI entrypoint for ai-api.
|
|
|
|
Reached at https://api.yokogawa.tech, behind Caddy and Authelia. There is no
|
|
authentication in this application because Authelia does it at the edge - which
|
|
also means this app must never be given a published host port, and never a
|
|
Caddyfile block without `import authelia`.
|
|
|
|
Every request is traced to Langfuse with its class, confidence, tool calls,
|
|
retrieved chunks, tokens, latency and contract result. Tracing failures never
|
|
break the request path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
import uuid
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel, Field
|
|
|
|
import agent
|
|
import documents
|
|
from config import settings
|
|
from contracts import ContractViolation
|
|
from guardrails import GuardrailViolation
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s %(message)s",
|
|
)
|
|
log = logging.getLogger("api")
|
|
|
|
app = FastAPI(
|
|
title="WRPS Plant Operations Assistant",
|
|
version="0.1.0",
|
|
description=(
|
|
"Information retrieval and analysis for the Waterloo Road Pump Station. "
|
|
"Not a control system, not an advisory controller, not a substitute for "
|
|
"a competent person."
|
|
),
|
|
)
|
|
|
|
# The operator's page is same-origin as of 2026-08-27, so it makes no
|
|
# cross-origin call and none of this applies to it. CORS is kept for the
|
|
# tunnelled build and for anything calling api.yokogawa.tech directly.
|
|
#
|
|
# NO_LLM_STUB additionally allows a tunnelled origin, because the whole chain
|
|
# has to be exercisable without Caddy and Authelia. It is deliberately tied to
|
|
# the stub flag so it disappears when the flag does.
|
|
#
|
|
# Earlier this file recorded the same-origin option as rejected, on the grounds
|
|
# that a second route to the API bypasses the api.yokogawa.tech Caddy block,
|
|
# where the Phase 9 publisher rule for ^/docs/.* lives. That objection was to
|
|
# proxying the WHOLE API through ai-web's nginx, and it still stands. What was
|
|
# built instead is narrower: Caddy routes ONLY /ask under ai.yokogawa.tech,
|
|
# through the same `import authelia` gate; /docs has no route there and 404s.
|
|
# The reason it had to change is that api.yokogawa.tech has no pinpoint DNS
|
|
# record, so an operator on cicore1 cannot resolve it at all. See
|
|
# caddy/ai-routes.caddy, which carries the do-not-widen warning.
|
|
_ORIGINS = ["https://ai.yokogawa.tech"]
|
|
if settings().no_llm_stub:
|
|
_ORIGINS.append("http://localhost:8080")
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=_ORIGINS,
|
|
# The UI sends credentials: "include" so the browser attaches the Authelia
|
|
# session cookie. Without this the preflight fails and NO cross-origin call
|
|
# succeeds - the browser refuses a credentialed request unless the response
|
|
# says Access-Control-Allow-Credentials: true. Found driving the real UI in
|
|
# a browser; it would have failed identically at ai.yokogawa.tech, and it
|
|
# cannot be found by calling the API directly with curl.
|
|
allow_credentials=True,
|
|
allow_methods=["GET", "POST"],
|
|
allow_headers=["Content-Type"],
|
|
)
|
|
|
|
|
|
def _langfuse():
|
|
"""Langfuse client, or None. Never let observability break the answer path."""
|
|
cfg = settings()
|
|
if not (cfg.langfuse_public_key and cfg.langfuse_secret_key):
|
|
return None
|
|
try:
|
|
from langfuse import Langfuse
|
|
|
|
return Langfuse(
|
|
public_key=cfg.langfuse_public_key,
|
|
secret_key=cfg.langfuse_secret_key,
|
|
host=cfg.langfuse_host,
|
|
)
|
|
except Exception:
|
|
log.exception("Langfuse unavailable - continuing untraced")
|
|
return None
|
|
|
|
|
|
# The document library screens. Mounted under /documents, NOT /docs - FastAPI's
|
|
# Swagger UI already owns /docs and the customer keeps it. Two things under one
|
|
# prefix with two different access policies is what gets misread later.
|
|
app.include_router(documents.router)
|
|
|
|
|
|
class AskRequest(BaseModel):
|
|
question: str = Field(min_length=3, max_length=1000)
|
|
question_class: str | None = Field(
|
|
default=None,
|
|
description=(
|
|
"Force the class instead of classifying. NO-LLM STUB MODE ONLY - "
|
|
"ignored unless NO_LLM_STUB is on, because letting a caller choose "
|
|
"its own contract is exactly what the classifier exists to prevent."
|
|
),
|
|
)
|
|
|
|
|
|
class AskResponse(BaseModel):
|
|
request_id: str
|
|
question_class: str
|
|
confidence: float
|
|
downgraded_reason: str | None = None
|
|
answer: dict
|
|
latency_ms: int
|
|
|
|
|
|
@app.get("/healthz")
|
|
def healthz() -> dict:
|
|
"""Liveness plus configuration shape. No secrets, ever - values are
|
|
reported as set/unset."""
|
|
return {"status": "ok", "config": settings().redacted()}
|
|
|
|
|
|
@app.post("/ask", response_model=AskResponse)
|
|
def ask(request: AskRequest) -> AskResponse:
|
|
request_id = str(uuid.uuid4())
|
|
started = time.perf_counter()
|
|
client = _langfuse()
|
|
trace = None
|
|
if client is not None:
|
|
try:
|
|
trace = client.trace(
|
|
id=request_id, name="ask", input={"question": request.question}
|
|
)
|
|
except Exception:
|
|
log.exception("could not open a Langfuse trace")
|
|
|
|
try:
|
|
answer, classification = agent.answer(
|
|
request.question, trace=trace, forced_class=request.question_class
|
|
)
|
|
except ContractViolation as violation:
|
|
# The offending output has already gone to Langfuse with the whole
|
|
# generated text. What comes back here says nothing about it: an error
|
|
# message is not a side channel for content that failed a safety check.
|
|
log.warning("contract failure request_id=%s rule=%s", request_id, violation.rule)
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail={
|
|
"request_id": request_id,
|
|
"error": "contract_not_met",
|
|
"message": (
|
|
"I could not produce an answer that meets the safety contract "
|
|
"for this question. The attempt has been logged for review."
|
|
),
|
|
},
|
|
) from violation
|
|
except GuardrailViolation as violation:
|
|
log.warning("guardrail refusal request_id=%s rule=%s", request_id, violation.rule)
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail={
|
|
"request_id": request_id,
|
|
"error": "query_refused",
|
|
"message": f"The query was refused: {violation.rule}.",
|
|
},
|
|
) from violation
|
|
except Exception as exc:
|
|
log.exception("unhandled error request_id=%s", request_id)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"request_id": request_id, "error": "internal_error"},
|
|
) from exc
|
|
|
|
latency_ms = int((time.perf_counter() - started) * 1000)
|
|
payload = answer.model_dump(mode="json")
|
|
|
|
if trace is not None:
|
|
try:
|
|
trace.update(
|
|
output=payload,
|
|
metadata={
|
|
"question_class": classification.question_class.value,
|
|
"confidence": classification.confidence,
|
|
"downgraded_reason": classification.downgraded_reason,
|
|
"contract_result": "passed",
|
|
"used_fixture_data": payload.get("used_fixture_data", False),
|
|
"latency_ms": latency_ms,
|
|
},
|
|
)
|
|
except Exception:
|
|
log.exception("could not finalise the Langfuse trace")
|
|
|
|
return AskResponse(
|
|
request_id=request_id,
|
|
question_class=classification.question_class.value,
|
|
confidence=classification.confidence,
|
|
downgraded_reason=classification.downgraded_reason,
|
|
answer=payload,
|
|
latency_ms=latency_ms,
|
|
)
|