What Is a One-Person AI Business?
In 2026, a one-person AI business is a solo-founded company where artificial intelligence handles the majority of operational tasks—customer support, content generation, code production, data analysis, and even decision-making—while you, the founder, focus on strategy, product vision, and high-level oversight. You aren't just using AI as a tool; you're architecting a system where multiple AI agents collaborate to deliver value to customers with minimal human intervention.
Think of it as building a digital organism. You define the skeleton (the business logic and data flows), wire up the nervous system (APIs and integrations), and train the brain (fine-tuned models and agentic workflows). Once deployed, the business runs, learns, and adapts largely on its own. Your role shifts from operator to curator—monitoring dashboards, approving major changes, and occasionally stepping in when edge cases slip past the AI's guardrails.
Why This Matters Now
The economics have fundamentally shifted. In 2024, running inference on GPT-4o cost roughly $0.01–$0.03 per 1K tokens. By early 2026, equivalent open-weight models like Llama 4, Mistral Large 2, and DeepSeek-R1 run on commodity hardware at a fraction of that cost, often with sub-100ms latency. Meanwhile, vertical AI models have matured to the point where they outperform general-purpose models on narrow business tasks. The result: a solo founder can now deploy an AI stack that would have required a 15-person engineering team just three years ago.
The competitive advantage isn't just cost savings—it's speed. A one-person AI business can ship features overnight, iterate on customer feedback within hours, and operate across time zones while you sleep. The barrier isn't technical anymore; it's knowing how to assemble the pieces correctly.
The Complete Stack: Architecture Overview
Let's walk through a production-ready stack for a hypothetical SaaS product: DocuMind AI—a tool that ingests company documents, answers questions via a chatbot, and generates weekly insight reports. This stack generalizes to virtually any AI-powered SaaS. Here's the full blueprint:
Layer 1: Foundation Models & Inference
Layer 2: Data Pipeline & Vector Storage
Layer 3: Agentic Orchestration
Layer 4: API Gateway & Backend
Layer 5: Frontend & User Experience
Layer 6: Observability, Billing & Deployment
We'll tackle each layer with real, runnable code examples.
Layer 1: Foundation Models & Inference
In 2026, you have three viable paths for model inference. Choose based on your latency requirements, data privacy needs, and budget.
- Self-hosted open-weight models (Llama 4, Mistral Large 2) via vLLM or TGI on a GPU cloud instance—best for data-sensitive workloads where you cannot send raw documents to third-party APIs.
- Managed inference providers (Groq, Fireworks, Together AI)—best for ultra-low latency with zero DevOps overhead. Groq routinely hits 300+ tokens/second on Llama 4 at ~$0.50 per million tokens.
- Fine-tuned vertical models hosted on replicate or custom endpoints—best when you need specialized behavior (e.g., legal document parsing) that general models mishandle.
For DocuMind AI, we'll use Groq for the customer-facing chatbot (needs sub-second latency) and a self-hosted Llama 4 via vLLM for the weekly insight reports that process large batches of sensitive documents overnight.
Setting Up vLLM with Llama 4 (Self-Hosted)
# Install vLLM on a Lambda Labs A100 instance (or RunPod, CoreWeave)
# Ubuntu 22.04 with CUDA 12.4+
# Step 1: Install dependencies
pip install vllm==0.6.0 transformers torch
# Step 2: Download Llama 4 (example: Meta-Llama-4-70B-Instruct)
# Via Hugging Face (ensure you've accepted the license)
huggingface-cli download meta-llama/Meta-Llama-4-70B-Instruct --local-dir /models/llama4-70b
# Step 3: Launch the vLLM server
python -m vllm.entrypoints.openai.api_server \
--model /models/llama4-70b \
--tensor-parallel-size 2 \
--max-model-len 8192 \
--gpu-memory-utilization 0.95 \
--port 8000 \
--host 0.0.0.0
# The server now accepts OpenAI-compatible requests at http://localhost:8000/v1
# Test with a curl:
curl http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"model": "meta-llama/Meta-Llama-4-70B-Instruct",
"messages": [{"role": "user", "content": "Summarize this document: ..."}],
"temperature": 0.2,
"max_tokens": 2048
}'
Connecting to Groq for Real-Time Chat
# requirements.txt
# groq==0.13.0
# httpx==0.27.0
import os
from groq import AsyncGroq
class GroqChatClient:
def __init__(self):
self.client = AsyncGroq(
api_key=os.environ.get("GROQ_API_KEY")
)
# Use Llama 4 70B on Groq for blazing fast inference
self.model = "llama-4-70b-versatile"
async def chat_completion(self, messages: list, temperature: float = 0.2):
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=1024,
top_p=1,
stream=False
)
return response.choices[0].message.content
async def stream_chat(self, messages: list):
"""Stream tokens for real-time UI updates."""
stream = await self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.2,
max_tokens=1024,
stream=True
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield delta
# Usage example
# client = GroqChatClient()
# response = await client.chat_completion([
# {"role": "system", "content": "You are a helpful document analyst."},
# {"role": "user", "content": "What are the key risks in the Q4 report?"}
# ])
Layer 2: Data Pipeline & Vector Storage
Your AI is only as good as the data it retrieves. In 2026, the dominant pattern is hybrid search—combining dense embeddings with sparse keyword retrieval (BM25) plus reranking. This gives you the semantic understanding of embeddings with the precision of lexical matching.
Document Ingestion Pipeline
# ingestion_pipeline.py
# Dependencies: langchain==0.3.0, chromadb==0.5.0, sentence-transformers==3.1.0
# PyMuPDF (fitz) for PDF parsing, unstructured for mixed document types
import hashlib
import json
from pathlib import Path
from typing import List, Dict
import fitz # PyMuPDF
from sentence_transformers import SentenceTransformer
import chromadb
from chromadb.config import Settings
class DocumentIngestionPipeline:
def __init__(self, chroma_host: str = "localhost", chroma_port: int = 8001):
# Embedding model - BGE-M3 is the 2026 gold standard for multilingual dense+sparse
self.embedding_model = SentenceTransformer(
"BAAI/bge-m3",
device="cuda"
)
# ChromaDB client (can also use LanceDB, Weaviate, or pgvector)
self.chroma_client = chromadb.HttpClient(
host=chroma_host,
port=chroma_port
)
self.collection = self.chroma_client.get_or_create_collection(
name="documents_v2",
metadata={"hnsw:space": "cosine"}
)
def extract_text_from_pdf(self, file_path: str) -> List[Dict]:
"""Extract text chunks with page numbers and section metadata."""
doc = fitz.open(file_path)
chunks = []
for page_num, page in enumerate(doc):
text = page.get_text(sort=True)
if not text.strip():
continue
# Split into paragraphs, filter noise
paragraphs = [p.strip() for p in text.split('\n\n') if len(p.strip()) > 50]
for para_idx, para in enumerate(paragraphs):
chunks.append({
"text": para,
"metadata": {
"source": Path(file_path).name,
"page": page_num + 1,
"paragraph_index": para_idx,
"chunk_hash": hashlib.md5(para.encode()).hexdigest()[:12]
}
})
return chunks
def create_embeddings_and_store(self, chunks: List[Dict]):
"""Embed all chunks and store in ChromaDB with dedup."""
texts = [chunk["text"] for chunk in chunks]
metadatas = [chunk["metadata"] for chunk in chunks]
ids = [f"{m['source']}_{m['page']}_{m['chunk_hash']}" for m in metadatas]
# Generate dense embeddings (BGE-M3 also produces sparse lexical weights)
embeddings = self.embedding_model.encode(
texts,
batch_size=32,
show_progress_bar=True,
normalize_embeddings=True
)
# Upsert in batches to avoid memory issues
BATCH_SIZE = 100
for i in range(0, len(ids), BATCH_SIZE):
batch_slice = slice(i, i + BATCH_SIZE)
self.collection.upsert(
ids=ids[batch_slice],
embeddings=embeddings[batch_slice].tolist(),
documents=texts[batch_slice],
metadatas=metadatas[batch_slice]
)
return len(ids)
# Usage:
# pipeline = DocumentIngestionPipeline()
# chunks = pipeline.extract_text_from_pdf("q4_report_2025.pdf")
# count = pipeline.create_embeddings_and_store(chunks)
# print(f"Ingested {count} chunks into vector store")
Hybrid Search with Reranking
# hybrid_search.py
# Dependencies: cohere==5.15.0, chromadb, sentence-transformers
import os
import cohere
from sentence_transformers import SentenceTransformer, CrossEncoder
from chromadb import HttpClient
from typing import List, Dict
class HybridSearchEngine:
def __init__(self):
self.embedding_model = SentenceTransformer("BAAI/bge-m3", device="cuda")
# Cross-encoder for reranking - Cohere Rerank v3 or local BGE-Reranker-v2-m3
self.reranker = CrossEncoder(
"BAAI/bge-reranker-v2-m3",
device="cuda",
max_length=1024
)
self.chroma = HttpClient(host="localhost", port=8001)
self.collection = self.chroma.get_collection("documents_v2")
# Optional: Cohere client for cloud-based reranking
self.co = cohere.ClientV2(api_key=os.getenv("CO_API_KEY"))
def dense_search(self, query: str, k: int = 20) -> List[Dict]:
"""Vector similarity search."""
query_embedding = self.embedding_model.encode(
query,
normalize_embeddings=True
).tolist()
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=k,
include=["documents", "metadatas", "distances"]
)
return [
{
"text": doc,
"metadata": meta,
"score": 1 - dist # Convert cosine distance to similarity
}
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
)
]
def rerank_results(self, query: str, candidates: List[Dict], top_k: int = 5) -> List[Dict]:
"""Rerank candidates using cross-encoder for precise relevance scoring."""
texts = [c["text"] for c in candidates]
# Cross-encoder scoring
pairs = [(query, text) for text in texts]
scores = self.reranker.predict(pairs)
# Attach scores and sort
for candidate, score in zip(candidates, scores):
candidate["rerank_score"] = float(score)
reranked = sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)
return reranked[:top_k]
def hybrid_search(self, query: str, top_k: int = 5) -> List[Dict]:
"""Full pipeline: dense retrieval -> rerank -> top_k results."""
# Step 1: Broad retrieval
candidates = self.dense_search(query, k=20)
# Step 2: Precision reranking
final_results = self.rerank_results(query, candidates, top_k=top_k)
return final_results
# Usage:
# searcher = HybridSearchEngine()
# results = searcher.hybrid_search("What were the Q4 revenue projections?", top_k=5)
# for r in results:
# print(f"[Score: {r['rerank_score']:.3f}] {r['text'][:200]}...")
Layer 3: Agentic Orchestration
This is where the magic happens. In 2026, agentic frameworks have matured beyond the experimental LangChain/LangGraph phase into production-grade systems. The key insight: you don't need a single monolithic agent. You need a swarm of narrow, reliable agents that communicate via structured outputs.
For DocuMind AI, we need three agents: a Query Agent (handles customer chat), a Report Agent (generates weekly insights), and a Guard Agent (validates outputs, prevents hallucination, enforces compliance).
Building a Reliable Agent Loop
# agent_orchestrator.py
# Using the OpenAI-compatible tool-calling pattern (works with vLLM, Groq, etc.)
import json
import asyncio
from dataclasses import dataclass, field
from typing import List, Dict, Any, Callable
from pydantic import BaseModel
import openai # Using the standard OpenAI client (works with vLLM/Groq endpoints)
@dataclass
class ToolDefinition:
name: str
description: str
parameters: Dict[str, Any]
handler: Callable
class QueryAgent:
"""Handles customer questions by searching documents and synthesizing answers."""
def __init__(self, search_engine, chat_client, model: str = "llama-4-70b"):
self.search = search_engine
self.chat = chat_client
self.model = model
# Define tools the agent can call
self.tools = [
{
"type": "function",
"function": {
"name": "search_documents",
"description": "Search the document store for relevant information",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
},
"top_k": {
"type": "integer",
"description": "Number of results",
"default": 5
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "get_document_context",
"description": "Fetch surrounding context from a specific document chunk",
"parameters": {
"type": "object",
"properties": {
"chunk_id": {"type": "string"},
"pages_before": {"type": "integer", "default": 1},
"pages_after": {"type": "integer", "default": 1}
},
"required": ["chunk_id"]
}
}
}
]
async def execute_tool(self, tool_name: str, arguments: Dict) -> str:
"""Execute a tool call and return the result as a string."""
if tool_name == "search_documents":
results = self.search.hybrid_search(arguments["query"], arguments.get("top_k", 5))
formatted = []
for i, r in enumerate(results):
formatted.append(
f"[Result {i+1} | Source: {r['metadata']['source']} Page {r['metadata']['page']}] "
f"{r['text']}"
)
return "\n\n".join(formatted)
elif tool_name == "get_document_context":
# Fetch surrounding chunks from ChromaDB
chunk_id = arguments["chunk_id"]
# Implementation: query ChromaDB for chunks with nearby page numbers
return f"Context around chunk {chunk_id}: [additional document text...]"
return "Tool not found"
async def run(self, user_query: str, conversation_history: List[Dict] = None) -> Dict:
"""Execute a full agent loop with tool calling."""
messages = [
{
"role": "system",
"content": (
"You are a precise document analyst. Answer questions using the provided tools. "
"Always cite sources with page numbers. If you're unsure, say so rather than guessing. "
"Use the search_documents tool before answering any factual question."
)
}
]
if conversation_history:
messages.extend(conversation_history[-10:]) # Keep context window manageable
messages.append({"role": "user", "content": user_query})
# Agent loop - allow up to 3 tool calls per turn
for iteration in range(3):
response = await self.chat.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tools,
tool_choice="auto",
temperature=0.1,
max_tokens=2048
)
message = response.choices[0].message
# If no tool calls, return the final answer
if not message.tool_calls:
return {
"answer": message.content,
"tool_calls_made": iteration,
"sources": self._extract_citations(message.content)
}
# Process tool calls
messages.append(message)
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
result = await self.execute_tool(tool_name, arguments)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# Fallback if max iterations reached
return {
"answer": "I've searched extensively but couldn't find a definitive answer. "
"Please rephrase your question or specify which document to search.",
"tool_calls_made": 3,
"sources": []
}
def _extract_citations(self, text: str) -> List[str]:
"""Parse citations from the answer text."""
import re
pattern = r'\[Source: ([^\]]+)\]'
return list(set(re.findall(pattern, text)))
The Guard Agent (Output Validation)
# guard_agent.py
# Validates outputs before they reach the customer
import re
from typing import Dict, Optional
class GuardAgent:
"""Validates AI outputs for safety, accuracy, and policy compliance."""
def __init__(self):
# Patterns to detect common failure modes
self.hallucination_markers = [
r"(?i)according to \[placeholder",
r"(?i)as mentioned in the document \[(?!.*\d)",
r"(?i)I don't have access to",
r"(?i)based on my training data",
]
self.forbidden_patterns = [
r"(?i)confidential internal memo",
r"(?i)personal identifiable information",
r"\b\d{3}-\d{2}-\d{4}\b", # SSN pattern
]
def validate(self, answer: str, retrieved_sources: list) -> Dict:
"""Run all validation checks."""
results = {
"passed": True,
"flags": [],
"confidence_score": 1.0
}
# Check 1: Hallucination detection
for pattern in self.hallucination_markers:
if re.search(pattern, answer):
results["flags"].append(f"Potential hallucination: {pattern}")
results["confidence_score"] -= 0.3
# Check 2: Forbidden content
for pattern in self.forbidden_patterns:
if re.search(pattern, answer):
results["flags"].append(f"Forbidden content detected: {pattern}")
results["passed"] = False
# Check 3: Source grounding - does the answer cite actual sources?
cited_sources = set()
source_pattern = r'\[Source: ([^\]]+)\]'
for match in re.finditer(source_pattern, answer):
cited_sources.add(match.group(1))
retrieved_source_names = set(
s.get('metadata', {}).get('source', 'unknown') for s in retrieved_sources
)
# If answer cites sources that weren't retrieved, flag it
ungrounded = cited_sources - retrieved_source_names
if ungrounded:
results["flags"].append(f"Ungrounded citations: {ungrounded}")
results["confidence_score"] -= 0.2
results["confidence_score"] = max(0.0, min(1.0, results["confidence_score"]))
return results
# Usage in the pipeline:
# guard = GuardAgent()
# validation = guard.validate(response["answer"], retrieved_sources)
# if not validation["passed"] or validation["confidence_score"] < 0.7:
# response["answer"] = "I wasn't able to find a reliable answer to your question. "
# "Please try rephrasing or contact support for assistance."
Report Agent (Batch Processing)
# report_agent.py
# Generates structured weekly reports by analyzing document clusters
import asyncio
from datetime import datetime
from typing import List, Dict
class ReportAgent:
"""Generates structured weekly insight reports from document collections."""
REPORT_SCHEMA = {
"type": "json_schema",
"json_schema": {
"name": "weekly_insight_report",
"strict": True,
"schema": {
"type": "object",
"properties": {
"executive_summary": {
"type": "string",
"description": "3-5 sentence overview of key findings"
},
"key_metrics": {
"type": "array",
"items": {
"type": "object",
"properties": {
"metric_name": {"type": "string"},
"current_value": {"type": "string"},
"trend": {
"type": "string",
"enum": ["up", "down", "stable"]
},
"source_page": {"type": "integer"}
},
"required": ["metric_name", "current_value", "trend"]
}
},
"risks_and_concerns": {
"type": "array",
"items": {
"type": "object",
"properties": {
"risk_description": {"type": "string"},
"severity": {
"type": "string",
"enum": ["critical", "high", "medium", "low"]
},
"mitigation_status": {"type": "string"}
}
}
},
"action_items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"item": {"type": "string"},
"owner": {"type": "string"},
"deadline": {"type": "string"}
}
}
}
},
"required": ["executive_summary", "key_metrics", "risks_and_concerns", "action_items"]
}
}
}
def __init__(self, search_engine, llm_client):
self.search = search_engine
self.llm = llm_client
async def generate_weekly_report(self, document_ids: List[str]) -> Dict:
"""Generate a structured weekly report from specified documents."""
# Step 1: Retrieve all chunks from target documents
all_context = []
for doc_id in document_ids:
results = self.search.collection.get(
where={"source": doc_id},
include=["documents", "metadatas"],
limit=200
)
for doc, meta in zip(results["documents"], results["metadatas"]):
all_context.append({
"text": doc,
"source": meta["source"],
"page": meta["page"]
})
# Step 2: Cluster into sections (financials, risks, product, etc.)
sections = self._cluster_by_topic(all_context)
# Step 3: Generate report sections in parallel
tasks = []
for section_name, chunks in sections.items():
context_text = "\n\n".join(
f"[{c['source']} p.{c['page']}] {c['text'][:500]}" for c in chunks[:15]
)
tasks.append(self._analyze_section(section_name, context_text))
section_results = await asyncio.gather(*tasks)
# Step 4: Synthesize final report
synthesis_prompt = f"""
You are a chief analyst. Synthesize the following section analyses into a cohesive report.
Today's date: {datetime.now().strftime('%Y-%m-%d')}
Section Analyses:
{json.dumps(section_results, indent=2)}
Generate a complete weekly insight report following the schema.
"""
final_report = await self.llm.chat_completion(
messages=[{"role": "user", "content": synthesis_prompt}],
temperature=0.1,
max_tokens=4096
)
return json.loads(final_report)
def _cluster_by_topic(self, chunks: List[Dict]) -> Dict[str, List[Dict]]:
"""Simple keyword-based clustering. Replace with embedding clustering in production."""
topics = {
"financials": ["revenue", "profit", "cost", "budget", "expense", "margin"],
"risks": ["risk", "threat", "vulnerability", "compliance", "regulatory"],
"product": ["feature", "release", "roadmap", "customer", "user", "bug"],
"operations": ["hiring", "team", "infrastructure", "vendor", "timeline"],
}
clusters = {topic: [] for topic in topics}
clusters["misc"] = []
for chunk in chunks:
text_lower = chunk["text"].lower()
matched = False
for topic, keywords in topics.items():
if any(kw in text_lower for kw in keywords):
clusters[topic].append(chunk)
matched = True
break
if not matched:
clusters["misc"].append(chunk)
return {k: v for k, v in clusters.items() if v}
async def _analyze_section(self, section_name: str, context: str) -> Dict:
"""Analyze a single section and return structured findings."""
prompt = f"""
Analyze the following document excerpts related to "{section_name}".
Context:
{context}
Extract:
1. Key metrics with values and trends
2. Any risks or concerns
3. Action items
Return valid JSON.
"""
result = await self.llm.chat_completion(
messages=[{"role": "user", "content": prompt}],
temperature=0.0
)
return {"section": section_name, "analysis": result}
Layer 4: API Gateway & Backend
In 2026, FastAPI remains the dominant Python backend framework, but the ecosystem has evolved. You now pair it with async-first database drivers (asyncpg for PostgreSQL, motor for MongoDB) and edge-deployed middleware via Cloudflare Workers for rate limiting and authentication.
FastAPI Backend with Agent Integration
# main_api.py
# Dependencies: fastapi==0.115.0, uvicorn==0.30.0, asyncpg==0.30.0, stripe==10.0.0
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import Optional, List
import asyncpg
import os
import json
app = FastAPI(title="DocuMind AI API", version="2.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=[os.getenv("FRONTEND_URL", "*")],
allow_methods=["*"],
allow_headers=["*"],
)
# Database pool
db_pool: Optional[asyncpg.Pool] = None
@app.on_event("startup")
async def startup():
global db_pool
db_pool = await asyncpg.create_pool(
dsn=os.getenv("DATABASE_URL"),
min_size=2,
max_size=10
)
# Initialize agents (lazy-loaded in production)
app.state.search_engine = HybridSearchEngine()
app.state.chat_client = GroqChatClient()
app.state.query_agent = QueryAgent(app.state.search_engine, app.state.chat_client)
app.state.guard_agent = GuardAgent()
@app.on_event("shutdown")
async def shutdown():
if db_pool:
await db_pool.close()
# --- Request/Response Models ---
class ChatRequest(BaseModel):
query: str
conversation_id: Optional[str] = None
document_filter: Optional[List[str]] = None # Limit search to specific docs
class ChatResponse(BaseModel):
answer: str
sources: List[dict]
conversation_id: str
confidence_score: float
class ReportRequest(BaseModel):
document_ids: List[str]
report_type: str = "weekly" # weekly, quarterly, custom
# --- API Endpoints ---
@app.post("/api/v2/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
"""Customer-facing chat with full agent pipeline."""
# Load conversation history if continuing
history = []
if request.conversation_id:
async with db_pool.acquire() as conn:
rows = await conn.fetch(
"SELECT role, content FROM messages WHERE conversation_id = $1 ORDER BY created_at LIMIT 20",
request.conversation_id
)
history = [{"role": row["role"], "content": row["content"]} for row in rows]
# Run query agent
result = await app.state.query_agent.run(request.query, history)
# Validate output
validation = app.state.guard_agent.validate(result["answer"], result.get("sources", []))
# Store conversation
conversation_id = request.conversation_id or f"conv_{os.urandom(8).hex()}"
async with db_pool.acquire() as conn:
await conn.execute(
"INSERT INTO messages (conversation_id, role, content) VALUES ($1, $2, $3)",
conversation_id, "user", request.query
)
await conn.execute(
"INSERT INTO messages (conversation_id, role, content) VALUES ($1, $2, $3)",
conversation_id, "assistant", result["answer"]
)
return {
"answer": result["answer"],
"sources": result.get("sources", []),
"conversation_id": conversation_id,
"confidence_score": validation["confidence_score"]
}
@app.post("/api/v2/reports/generate")
async def generate_report(request: ReportRequest, background_tasks: BackgroundTasks):
"""Trigger report generation (async, returns job ID)."""
job_id = f"rpt_{os.urandom(6).hex()}"
# Store job in DB
async with db_pool.acquire() as conn:
await conn.execute(
"INSERT INTO report_jobs (job_id, status, document_ids, created_at) VALUES ($1, $2, $3, NOW())",
job_id, "processing", json.dumps(request.document_ids)
)
# Fire-and-forget background task
background_tasks.add_task(_generate_report_background, job_id, request.document_ids)
return {"job_id": job_id, "status": "processing", "estimated_seconds": 45}
async def _generate_report_background(job_id: str, document_ids: List[str]):
"""Background report generation."""
try:
report_agent = ReportAgent(app.state.search_engine, app.state.chat_client)
report = await report_agent.generate_weekly_report(document_ids)
async with db_pool.acquire() as conn:
await conn.execute(
"UPDATE report_jobs SET status = $1, result = $2, completed_at = NOW() WHERE job_id = $3",
"completed", json.dumps(report), job_id
)
except Exception as e:
async with db_pool.acquire() as conn:
await conn.execute(
"UPDATE report_jobs SET status = $1, error = $2 WHERE job_id = $3",
"failed", str(e), job_id
)