Python: document assistant example
Instrument an ASGI application and add retrieval, explanation and source observations.
#1. Install the wheel
Use Python 3.10+ and the service-specific wheel downloaded from the portal.
python -m pip install ./packages/sigillum_finans_sdk-0.11.0-py3-none-any.whl
python -m pip install fastapi uvicornHTTPX, OpenTelemetry and cryptography dependencies can be resolved through your institution’s package mirror.
"""Deterministic document retrieval; no live LLM call."""
import json
import os
from pathlib import Path
from fastapi import FastAPI
from pydantic import BaseModel, Field
from sigillum_finans import start, JsonData, Attachment, References
sdk = start(Path(os.environ["SIGILLUM_CONFIG"]).resolve())
api = FastAPI(title="Sigillum SDK · Document assistant example")
DOCUMENTS = (
("D01", "Apply for a card through the mobile application."),
("D02", "Update your address from the customer profile screen."),
("D03", "Make a transfer from the payments menu."),
)
def data(value):
return JsonData.raw(json.dumps(value, ensure_ascii=False, separators=(",", ":")))
class Query(BaseModel):
request_ref: str = Field(min_length=1, max_length=64)
question: str = Field(min_length=1, max_length=500)
@api.post("/assist")
async def assist(query: Query):
sdk.set_references(References(session_ref=query.request_ref))
with sdk.start_step("document-retrieval") as step:
words = set(query.question.casefold().split())
ranked = sorted(DOCUMENTS,
key=lambda item: len(words & set(item[1].casefold().split())), reverse=True)
chosen = ranked[0]
step.observe("ai", "retrieval", data({
"method": "synthetic-keyword-overlap",
"query": query.question, "documents": [{"id": chosen[0], "text": chosen[1]}]
}))
with sdk.start_step("answer-compose") as step:
result = {"request_ref": query.request_ref, "answer": chosen[1],
"source": chosen[0], "demonstration": True}
step.observe("ai", "explanation", data({
"method": "document-quotation", "source": chosen[0],
"message": "Selected the example document with the highest word overlap."
}))
step.observe("custom", "assistant.result", data(result))
original = json.dumps(result, ensure_ascii=False).encode("utf-8")
step.observe("source", "comparison.source", data({"reference": query.request_ref}),
attachments=(Attachment("result", original, "application/json",
"comparison.originalContent"),))
return result
@api.get("/sdk-status")
def status():
return {"ready": sdk.status.state == "ready", "pending_records": sdk.status.pending_records,
"lost": sdk.status.lost}
app = sdk.wrap_asgi(api)
#2. Wrap the ASGI application
Create a single SDK with start(Path(...).resolve()). After defining the routes, use app = sdk.wrap_asgi(api). The SDK observes standard ASGI HTTP messages and in-scope HTTPX calls. WebSockets, arbitrary sockets and every Python library are not automatically instrumented.
#3. Record the document-assistant workflow
The POST /assist example accepts a question, selects from three synthetic local documents by word matching and produces a deterministic answer. It does not call a real LLM. Retrieval source content, explanation, business result and original source attachment are recorded explicitly. Other AI categories are demonstrated separately in the AI observation examples.
#4. Supply JSON, attachments and references
JsonData.raw accepts a valid JSON object as str or immutable bytes. Attachment carries original content. References and ExternalReference add person, session and business references. Avoid parsing and serialising large JSON numbers solely for observation.
#5. Integrate an actual model call
sdk.call_ai and call_ai_async execute the actual operation once. If the return value is not raw JSON, an output= selector supplies the original JSON representation. Do not pre-consume a stream for capture; observe it during normal consumption.
#6. Run the project
Follow the ZIP README and set SIGILLUM_CONFIG to the absolute service-configuration path. Start with uvicorn app:app --host 127.0.0.1 --port 5110. Set POST /assist as the entry point. A /sdk-status response is not evidence of transaction observation: send an actual /assist request.