Alkiviades — AI Pipeline Reference¶
The complete generative AI pipeline from user query to final response. Documents prompts, system prompts, workflow state machines, and the SSE streaming architecture.
Overview¶
User Query
│
▼
┌─────────────────────────────────────────────────────┐
│ 1. RAG RETRIEVAL │
│ Embedder (TF-IDF) → cosine similarity → top-5 │
│ chunks from 2,461 documents │
└──────────────────┬──────────────────────────────────┘
│ chunks + scores
▼
┌─────────────────────────────────────────────────────┐
│ 2. PROMPT ASSEMBLY │
│ System prompt (dynamic axes) + history (trimmed) │
│ + user message (with chunk context) │
└──────────────────┬──────────────────────────────────┘
│ messages[]
▼
┌─────────────────────────────────────────────────────┐
│ 3. TOOL-CALLING LOOP (max 5 iterations) │
│ ┌──────────────────────────────────────────────┐ │
│ │ LLM Stream (DeepSeek v4 Pro) │ │
│ │ ├─ TokenEvent → yield to SSE │ │
│ │ └─ finish_reason="tool_calls" → flag │ │
│ └──────────────┬───────────────────────────────┘ │
│ │ │
│ ┌─────────▼──────────┐ │
│ │ had_tool_call? │ │
│ └──────┬──────┬──────┘ │
│ Yes│ │No │
│ ┌───────▼──┐ │ │
│ │ Resolve │ │ │
│ │ tool calls│ │ │
│ │ (retry) │ │ │
│ └───────┬──┘ │ │
│ ┌───────▼──┐ │ │
│ │ Execute │ │ │
│ │ tool(s) │ │ │
│ │ (thread │ │ │
│ │ pool) │ │ │
│ └───────┬──┘ │ │
│ │ │ │
│ continue append to history │
│ loop yield DoneEvent │
│ trim if needed │
└──────────────────┬──────────────────────────────────┘
│ final answer
▼
┌─────────────────────────────────────────────────────┐
│ 4. PERSISTENCE │
│ Save turn to SQLite (prompt, response, sources, │
│ tool calls, chunks) │
└─────────────────────────────────────────────────────┘
1. System Prompt Assembly¶
Source: backend/app/prompt.txt + backend/app/config.py¶
The system prompt is built dynamically at runtime by build_system_prompt().
Prompt Template (prompt.txt)¶
You are Alkiviades, an AI consultant specializing in marketing,
strategic communication, and political campaigns. You answer based
on the provided knowledge base.
Your operating principles:
{ethical_description}
{alignment_description}
{social_description}
When answering:
1. Analyze relevant information from the context provided below.
2. Provide consultation within these principles.
3. Cite specific source files using [source: filename.md].
4. Respond in the language the user used.
Dynamic Axis Descriptions¶
_axis_description() snaps each axis value to the nearest discrete position (7 positions: -1.0, -0.67, -0.33, 0.0, +0.33, +0.67, +1.0) and generates the corresponding description block.
Ethical axis (-1 = unscrupulous through +1 = principled):
| Value | Label | Behavior |
|---|---|---|
| -1.0 | Unscrupulous | No ethical restraint. Deception and manipulation are acceptable. |
| -0.67 | Pragmatic | Results matter more than principles. |
| -0.33 | Flexible | Generally pragmatic, rare ethical exceptions. |
| 0.0 | Balanced | Weigh ethics alongside practical outcomes. |
| +0.33 | Principled | Prioritize ethical conduct, rare pragmatic exceptions. |
| +0.67 | Ethical | Strong commitment to principles. |
| +1.0 | Absolute adherence | Never compromise principles. Truth above all. |
Alignment axis (-1 = left/socialist through +1 = right/capitalist):
| Value | Label | Behavior |
|---|---|---|
| -1.0 | Left (socialist) | Collective ownership, worker rights, critical of capitalism. |
| -0.67 | Center-left | Worker protections, public services, progressive taxation. |
| -0.33 | Slight left lean | Social welfare, regulated markets. |
| 0.0 | Centrist | Mixed economy, pragmatic. |
| +0.33 | Slight right lean | Pro-market, targeted welfare. |
| +0.67 | Center-right | Private enterprise, moderate regulation. |
| +1.0 | Right (capitalist) | Free markets, private enterprise, low taxation. |
Social freedom axis (-1 = authoritarian through +1 = libertarian):
| Value | Label | Behavior |
|---|---|---|
| -1.0 | Authoritarian | State-centric, order and collective control. |
| -0.67 | Leaning authoritarian | Strong institutions, limited personal freedom. |
| -0.33 | Slight authoritarian lean | Order with some individual freedom. |
| 0.0 | Balanced | Individual freedom vs. social order. |
| +0.33 | Slight libertarian lean | Individual choice with state regulation. |
| +0.67 | Leaning libertarian | Personal autonomy, minimal state intervention. |
| +1.0 | Libertarian | Maximize individual freedom, skeptical of state overreach. |
Tool Hint (appended automatically)¶
You have access to these tools:
- search_news(query): Search for news and current information using Tavily.
- load_search(search_id): Load a previously saved search result.
- scrape_url(url): Fetch the content of a web page using ScrapingAnt.
- read_file(path): Read the contents of a file in the project.
- write_file(path, content): Create or overwrite a file.
- edit_file(path, old_text, new_text): Find and replace text in a file.
- append_file(path, content): Append text to an existing file.
When you receive tool results, analyze them and incorporate them
into your response. Always cite sources from both the knowledge base
[source: filename.md] and web searches [source: url].
Template rendering¶
build_system_prompt() in config.py:
1. Loads the prompt template from prompt.txt
2. Snaps each axis value to the nearest discrete position
3. Replaces {ethical_description}, {alignment_description}, {social_description} with dynamically generated blocks
4. Appends TOOL_HINT at the end
5. Returns the complete system prompt string
2. RAG Retrieval Pipeline¶
Ingestion (build time)¶
party_docs/ processed/ stratcom/
(HTML/PDF, ~120 files) (MD, 100 files) (MD, 9 files)
│ │ │
└─ preprocess.py ────┘ │
(BeautifulSoup ───────────────────┘
/markdownify) │
▼
KnowledgeBase.build()
│
┌──────────┴──────────┐
│ 1. Chunk MD into │
│ sections + paras │
│ (800 chars, 200 │
│ overlap) │
│ 2. Build TF-IDF vocab│
│ (top 15K terms) │
│ 3. Vectorize chunks │
│ (float16 array) │
│ 4. Save to pickle │
└──────────┬──────────┘
▼
.chroma_db/vectors.pkl
(2,461 chunks, ~75 MB)
Retrieval (query time)¶
# From utils.py — rag_retrieve()
def rag_retrieve(embedder, kb, question, top_k=5):
kb.load() # Load KB if not already
query_vec = embedder.embed_text(question) # TF-IDF vectorize
results = kb.query(query_vec, top_k) # Cosine similarity
return results # [ChunkResult, ...]
TF-IDF Vectorization: Pure numpy, zero ML dependencies. Language-agnostic (works for Greek + English). Configurable: EMBED_DTYPE (float16/float32), MAX_VOCAB_SIZE (default 15K terms).
Cosine Similarity: numpy.dot(query_vec, doc_vec) / (||query|| * ||doc||)
Memory: float16 + 15K vocab cap = ~75 MB. mmap mode for lazy loading on Render free tier (512 MB).
3. Tool-Calling Loop State Machine¶
Flow (backend/app/agent.py — process_stream())¶
┌─────────────────────────────────────────────────────────┐
│ process_stream(query, axes) │
│ │
│ 1. RAG RETRIEVE │
│ chunks = rag_retrieve(embedder, kb, query) │
│ for each: yield SourceEvent(file, score) │
│ │
│ 2. BUILD MESSAGES │
│ system = build_system_prompt(axes or self._axes) │
│ user_msg = build_user_message(query, chunks) │
│ messages = [system, *history, user_msg] │
│ │
│ 3. TOOL LOOP (for iteration in range(5)): │
│ ┌─────────────────────────────────────────────────┐ │
│ │ is_last = (iteration == 4) │ │
│ │ tools = None if is_last else TOOL_SCHEMAS │ │
│ │ │ │
│ │ if is_last: │ │
│ │ messages += "Provide final response..." │ │
│ │ │ │
│ │ streamed_text = "" │ │
│ │ for event in call_llm_stream(messages, tools): │ │
│ │ TokenEvent: streamed_text += text, yield │ │
│ │ ToolCallEvent: had_tool_call = True │ │
│ │ │ │
│ │ if had_tool_call and not is_last: │ │
│ │ tool_calls = resolve_tool_calls(messages) │ │
│ │ if tool_calls: │ │
│ │ for each tc: │ │
│ │ args = json.loads(tc.arguments) │ │
│ │ result = execute_tool(name, args) [thread]│ │
│ │ messages += tool result │ │
│ │ yield ToolCallEvent/ToolResultEvent │ │
│ │ continue # next iteration │ │
│ │ │ │
│ │ // No tool calls — FINAL ANSWER │ │
│ │ history += (user_msg, assistant(streamed_text)) │ │
│ │ trim_history() │ │
│ │ yield DoneEvent │ │
│ │ return │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ FALLTHROUGH: yield ErrorEvent("max iterations") │
└─────────────────────────────────────────────────────────┘
Tool Call Resolution¶
When finish_reason="tool_calls", the streaming response has no text content — only tool call IDs/names that need to be resolved. DeepSeek's streaming API sometimes returns incomplete tool call data, so a non-streaming retry is used:
# From utils.py — _resolve_tool_calls()
async def _resolve_tool_calls(messages, tools, api_key):
# Non-streaming retry to get complete tool call data
response = await call_llm_non_streaming(messages, tools, api_key)
return response.choices[0].message.tool_calls
Tool Execution¶
All tool execution happens in a thread pool (run_in_executor) to avoid blocking the async event loop:
4. Available Tools (backend/app/tools.py)¶
| Tool | Args | Description |
|---|---|---|
search_news |
query: str, include_domains?: list[str] |
Tavily API search. Saves results to searches/{id}.md. Returns markdown with title, URL, snippet. |
load_search |
search_id: str |
Load a previously saved search result. |
scrape_url |
url: str |
Fetch page content via ScrapingAnt. Returns markdown. |
read_file |
path: str |
Read file content. Path-restricted to project root. |
write_file |
path: str, content: str |
Create or overwrite file. Restricted to sessions/ workspace. |
edit_file |
path: str, old_text: str, new_text: str |
Find and replace in file. Restricted to sessions/. |
append_file |
path: str, content: str |
Append to file. Restricted to sessions/. |
5. SSE Streaming Architecture¶
Event Types¶
| SSE Event | Internal Event | Data |
|---|---|---|
token |
TokenEvent |
{"text": "..."} — single content delta |
source |
SourceEvent |
{"file": "...", "score": 0.XX} — RAG source |
tool_call |
ToolCallEvent |
{"tool": "...", "args": {...}} — tool invoked |
tool_result |
ToolResultEvent |
{"tool": "...", "result": "..."} — tool output |
done |
DoneEvent |
{"session_id": "..."} — stream complete |
error |
ErrorEvent |
{"message": "..."} — error occurred |
heartbeat |
(internal) | {"t": 1234567890} — keepalive every 15s |
Stream Orchestration (backend/app/main.py — chat/stream)¶
┌─────────────────────────────────────────────────┐
│ Event Source Response │
│ │
│ ┌─────────────────┐ ┌───────────────────┐ │
│ │ _produce_agent │ │ _produce_heartbeat │ │
│ │ _events() │ │ () │ │
│ │ │ │ │ │
│ │ agent.process_ │ │ every 15s: │ │
│ │ stream(query) │ │ queue.put( │ │
│ │ │ │ │ heartbeat) │ │
│ │ agent_to_sse() │ │ │ │
│ │ │ │ └────────┬──────────┘ │
│ │ queue.put(sse) │ │ │
│ │ │ │ │ │
│ │ collect answer/ │ │ │
│ │ sources/tools │ │ │
│ │ │ │ │ │
│ └────────┼─────────┘ │ │
│ │ │ │
│ ┌────▼───────────────────────▼──┐ │
│ │ asyncio.Queue │ │
│ │ (max 256) │ │
│ └────────────┬───────────────────┘ │
│ │ │
│ ┌────▼────────────┐ │
│ │ Main loop: │ │
│ │ queue.get(30s) │ │
│ │ ↓ │ │
│ │ Event is None? │ │
│ │ Yes → break │ │
│ │ No → yield SSE │ │
│ └────┬─────────────┘ │
│ │ │
│ _save_turn() ← persist to SQLite │
└─────────────────────────────────────────────────┘
Concurrent tasks:¶
_produce_agent_events()— runs agent loop, converts events to SSE, feeds queue_produce_heartbeat()— sends keepalive every 15s to prevent Render timeout- Main loop — drains queue and yields SSE events to client
On completion: _save_turn() writes the turn to SQLite (sessions + messages tables).
6. History Management¶
Storage (agent.py — _history)¶
self._history: list[dict] = []
# Each entry is a {"role": "user"|"assistant", "content": "..."} dict
After each successful response, both the user message and the full assistant response are appended to _history. This preserves full conversation context for subsequent turns, enabling multi-turn dialogue, follow-up questions, and contextual awareness.
Trimming (agent.py — _trim_history())¶
When the total token count (system prompt + history) approaches MAX_CONTEXT_TOKENS (48,000), older messages are summarized:
Token count = count_messages_tokens(system_prompt + history)
Available = MAX_CONTEXT_TOKENS - system_tokens - 4096 (safety margin)
if history_tokens > available:
1. Keep last SUMMARY_KEEP_LAST_N (6) messages intact
2. Send older messages to LLM for summarization:
"Summarize the following conversation history in 2-3 sentences..."
3. Replace old history with single system message:
{"role": "system", "content": "[Previous conversation summary: ...]"}
4. Append kept messages after summary
Fallback: if summarization fails (API error), simply drop older messages
7. Session Persistence¶
SQLite Schema¶
-- sessions table
CREATE TABLE sessions (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
title TEXT,
axes TEXT, -- JSON: {"ethical": 1.0, ...}
created_at TEXT,
updated_at TEXT,
FOREIGN KEY (user_id) REFERENCES users(uuid)
);
-- messages table
CREATE TABLE messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
turn_number INTEGER,
role TEXT, -- 'user' | 'assistant' | 'system'
content TEXT,
prompt TEXT, -- user's question (raw, for user messages)
response TEXT, -- assistant's answer (for assistant messages)
tools_called TEXT, -- JSON: [{"name": "search_news", "args": {...}}]
cited_sources TEXT, -- JSON: [{"file": "doc.md"}]
chunks TEXT, -- JSON: [{"source": "doc.md", "score": 0.85}]
timestamp TEXT,
FOREIGN KEY (session_id) REFERENCES sessions(id)
);
Turn Save Flow¶
- SSE stream completes (DoneEvent)
_save_turn()inmain.py:- Creates session if new (UUID, title from first message)
- Inserts user message (role=user, prompt=question)
- Inserts assistant message (role=assistant, response=answer)
- Both rows get tools_called, chunks, sources metadata
8. API Endpoints¶
Chat¶
| Method | Path | Description |
|---|---|---|
| POST | /chat |
Non-streaming chat (fallback). Returns {"reply": "..."} |
| POST | /chat/stream |
SSE streaming chat. Produces token, source, tool, done, error, heartbeat events |
Axes¶
| Method | Path | Description |
|---|---|---|
| GET | /api/axes |
Get current global axis values |
| PUT | /api/axes |
Set axis values (body: {"ethical": 0.5, ...}, clamped to [-1,1]) |
Sessions¶
| Method | Path | Description |
|---|---|---|
| GET | /sessions |
List user's sessions |
| GET | /sessions/{id} |
Get session (with ?preview=true for first 3 turns) |
| DELETE | /sessions/{id} |
Delete session |
9. LLM Parameters¶
| Parameter | Value | Config |
|---|---|---|
| Model | deepseek-v4-pro |
config.DEEPSEEK_MODEL |
| Temperature | 0.7 | config.LLM_TEMPERATURE |
| Max tokens | 4096 | config.LLM_MAX_TOKENS |
| Timeout | 120s | config.LLM_TIMEOUT |
| Max tool iterations | 5 | config.MAX_TOOL_ITERATIONS |
| Max context tokens | 48000 | config.MAX_CONTEXT_TOKENS |
| Summary keep last N | 6 | config.SUMMARY_KEEP_LAST_N |
| Chunk size | 800 | config.CHUNK_SIZE |
| Chunk overlap | 200 | config.CHUNK_OVERLAP |
| Top-K retrieval | 5 | config.TOP_K |