A small FastAPI backend around an LLM (DeepSeek, OpenAI-compatible Responses API): a tool-calling loop that queries an external RAG service, streamed answers, and a structured-output endpoint validated with Pydantic.
- Tool calling loop: the model can call
search_study_materialsseveral times, including several calls in one turn, up toMAX_TOOL_ROUNDSrounds per request. - Tool errors go back to the model: invalid arguments (Pydantic validation), an
unknown tool name or a failed RAG request become a
{"error": ...}tool result, so the model can explain the problem instead of the request crashing. - Streaming: after a tool round the answer is streamed as text deltas. A direct answer (no tool needed) comes from a single non-streamed call and is sent as one chunk.
- Structured output:
POST /routeasks the model for aRouteDecision(JSON Schema from Pydantic) and validates the reply; malformed output is a 502. - Provider failures: client timeout and retries are configurable; if the LLM is
unreachable,
/chatand/routereturn 503 before any output is sent. - Offline test suite (28 tests) and CI.
POST /chat ─► DeepSeek (tools offered)
│
├── text ─────────────────────────────► answer (one chunk)
│
└── function call(s)
│
▼
validate args ─► search_study_materials ─► manual-rag /ask
│ (errors become tool results)
▼
DeepSeek, streamed ─► text deltas ─► client
│
└── another function call? repeat, up to MAX_TOOL_ROUNDS
POST /route ─► DeepSeek with json_schema format ─► RouteDecision (Pydantic)
The RAG service is the separate manual-rag
project. Its /ask endpoint already generates an answer with an LLM, so a tool call
costs a second LLM round trip; a retrieval-only endpoint would be cheaper.
app/
├── main.py # FastAPI: /chat, /route, /health
├── llm.py # client, tool-calling loop, structured route decision
├── tools.py # search_study_materials and its tool definition
├── schemas.py # request models, tool arguments, RouteDecision
└── config.py # settings from environment variables
examples/
├── tool_use.py # step-by-step tool calling script with printed intermediate steps
└── structured.py # standalone structured output script
tests/ # offline pytest suite
Set in .env (see .env.example) or the environment:
| Variable | Default | Meaning |
|---|---|---|
DEEPSEEK_API_KEY |
— (required) | API key |
LLM_BASE_URL |
https://api.deepseek.com |
OpenAI-compatible endpoint |
LLM_MODEL |
deepseek-flash |
model name |
LLM_TIMEOUT / LLM_MAX_RETRIES |
30 / 2 |
client timeout (s) and retries |
RAG_URL / RAG_TIMEOUT |
http://127.0.0.1:8000/ask / 60 |
manual-rag endpoint and timeout (s) |
MAX_TOOL_ROUNDS |
3 |
tool rounds per /chat request |
python -m venv .venv
source .venv/Scripts/activate # Windows Git Bash; .venv/bin/activate on Linux/macOS
pip install -r requirements.txt
cp .env.example .env # then set DEEPSEEK_API_KEYStart manual-rag on port 8000, then this API:
uvicorn app.main:app --reload --port 8001curl -N http://127.0.0.1:8001/chat --json '{"message":"What is a derivative?"}'The response is plain text, streamed. message must be 1–4000 characters (422 otherwise).
curl http://127.0.0.1:8001/route --json '{"message":"What is a derivative?"}'{"use_rag": true, "search_query": "derivative definition", "reason": "..."}Returns {"status": "ok"}.
pip install -r requirements-dev.txt
pytestThe suite runs offline: the LLM client and the manual-rag HTTP call are replaced with fakes. It covers the direct answer, the tool round trip (each request replays the full tool history), parallel calls, a second tool round and the round limit, tool errors for invalid arguments, unknown tools and RAG failures, streaming of text deltas only, the structured route decision, input validation, and the 502/503 mappings. It does not call the real DeepSeek API.
On 2026-10-19 the service code was run against the real DeepSeek API (deepseek-flash
through the Responses API) with manual-rag deliberately offline:
- a greeting was answered directly, without a tool call;
- a study question triggered
search_study_materials; the RAG request failed, the model received{"error": "study materials search failed: ConnectError"}and streamed an answer (77 text deltas) saying the knowledge base was unavailable instead of inventing one; route_questionreturned a validRouteDecision(use_rag=truewith a search query).
Not exercised live: a second tool round within one request (covered by the offline tests).
Python · FastAPI · Pydantic · OpenAI Python SDK (Responses API) · DeepSeek · HTTPX · pytest