class AppResources:
    retriever: object              # Hybrid retriever (Milvus + ES)
    generator: object              # LLM handler
    answer_service: AnswerService  # Core RAG service
    user_chains: Dict              # Per-user conversation chains
    session_memory: SessionMemory  # User session memory
    thread_memory: ThreadMemory    # Conversation thread memory
    ingester: DocumentIngester     # Document ingestion pipeline
    org_manager: OrgResourceManager # Multi-tenant org isolation

# Example: Getting org-specific resources
org_res = resources.org_manager.get(org_id, init_ingester=True)
retriever = org_res["retriever"]
answer_service = org_res["answer_service"]

http://localhost:8002

GET /healthz

{
  "status": "ok"
}

POST /query

Authorization: Bearer <JWT_TOKEN>
Content-Type: application/json

{
  "text": "I need to review the compliance procedures before contacting the client",
  "user_info": {
    "user_name": "john@company.com",
    "real_name": "John Doe"
  }
}

{
  "intent": "compliance_check",
  "confidence": 0.95,
  "requires_rag": true,
  "input": "I need to review...",
  "user_info": {
    "user_name": "john@company.com",
    "real_name": "John Doe",
    "org_id": "uuid-here",
    "current_roles": ["analyst"],
    "current_groups": []
  }
}

POST /predict

Authorization: Bearer <JWT_TOKEN>
Content-Type: application/json

{
  "query": "What are the security implementation challenges in Olis?",
  "user_info": {
    "user_name": "developer@olis.com",
    "real_name": "Dev User"
  },
  "thread_id": "optional-existing-thread-id"
}

{
  "answer": {
    "callout_answer": "The main security challenges in Olis include...",
    "positive_list": [
      "Implementing end-to-end encryption for sensitive data",
      "Establishing secure authentication flows"
    ],
    "negative_list": [],
    "info_list": [
      "Security audit scheduled for Q2",
      "RBAC implementation in progress"
    ],
    "keywords": ["security", "encryption", "authentication", "RBAC"]
  },
  "sources": {
    "doc_1": {
      "title": "Security Implementation Plan",
      "source": "internal-wiki",
      "page_content": "...",
      "metadata": { "owner_id": "admin@olis.com", "acl": "|user:_all_|" }
    }
  },
  "query": "What are the security implementation challenges in Olis?",
  "original_query": "What are the security implementation challenges in Olis?",
  "input": "What are the security implementation challenges in Olis?",
  "thread_id": "uuid-thread-id"
}

POST /upsert

Authorization: Bearer <JWT_TOKEN>
Content-Type: application/json

{
  "user_email": "uploader@olis.com",
  "documents": {
    "doc_id_1": {
      "title": "Product Requirements Document",
      "content": "Full document text here...",
      "metadata": {
        "source": "confluence",
        "created_at": "2026-01-15",
        "owner_id": "pm@olis.com",
        "acl": "|user:_all_|role:pm|role:engineering|"
      }
    },
    "doc_id_2": {
      "title": "API Design Spec",
      "content": "...",
      "metadata": { "source": "github", "acl": "|role:engineering|" }
    }
  }
}

{
  "status": "success",
  "user": "uploader@olis.com",
  "num_chunks": "42"
}

POST /testpredict

Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

{
  "email": "user@company.com",
  "role": "engineer",
  "org_id": "uuid-org-id",
  "type": "access",
  "exp": 1738425600,
  "iat": 1738422000,
  "iss": "oauth-app",
  "jti": "unique-token-id"
}

JWT_SECRET=your-secret-key-here
JWT_ISSUER=oauth-app
JWT_ALGORITHM=HS256

|user:john@company.com|role:engineer|role:pm|group:product|deny:banned@company.com|

def _doc_allowed_for_user(doc: Any, user_info: Dict[str, Any]) -> bool:
    meta = doc.metadata
    user = user_info.get("user_name").lower()
    owner = meta.get("owner_id", "").lower()
    acl = meta.get("acl", "").lower()

    # Deny check
    if f"|deny:{user}|" in acl:
        return False

    # Owner check
    if owner == user:
        return True

    # Public check
    if "|user:_all_|" in acl:
        return True

    # User/role/group checks...
    # (See main.py:_doc_allowed_for_user for full implementation)

query_text = _normalize_query_value(input.query)
# Handles string or dict input, extracts query value

# Weighted combination
score = (vector_score * WEIGHT_VECTOR) + (bm25_score * WEIGHT_BM25)
# Default weights: 0.5 / 0.5

retrieval_ms = int((time.monotonic() - t0) * 1000)
_perf_emit("retrieval_done", {"retrieval_ms": retrieval_ms, "docs_count": len(docs)})

docs = _filter_docs_by_rbac(docs, user_info)
# Remove docs user doesn't have permission to access

context = answer_service.prepare_context(
    inputs={
        "retrieved_docs": docs,
        "query": query_text,
        "original_query": query_text,
        "subqueries": [query_text],
    },
    user_info=user_info,
    chat_history=None,  # Added from thread memory if available
)

raw_answer = answer_service.generate_answer_with_prepared_context(
    context=context,
    answer_prompt=AnswerWithKeywordsPrompt,
)

You are Olis, an intelligent assistant that answers questions using provided documents.

Context:
[Document chunks with [Source: doc_1] markers]

Question: {query}

Instructions:
- Provide a comprehensive answer using the context
- Include specific examples and details
- Cite sources using [Source: doc_id] format
- Structure your answer with:
  - callout_answer: Main answer summary
  - positive_list: Specific points/facts supporting the answer
  - negative_list: Caveats/limitations
  - info_list: Additional relevant information
  - keywords: Key terms from the answer

Format your response as JSON.

# Flatten answer + keywords structure
payload = answer_service.flatten_ak(context)

# Coalesce citation keys ([Source: doc_1] -> sources dict)
payload = answer_service.coalesce_citation_keys(payload)

# Remove unused sources not cited in answer
payload = answer_service.prune_sources_by_citations(payload)

# Parse JSON string answers to objects
payload = _maybe_parse_answer(payload)

{
  "answer": {
    "callout_answer": "No information could be found for the question based on the provided sources.",
    "positive_list": [],
    "negative_list": [],
    "info_list": [],
    "keywords": []
  },
  "sources": {},
  "query": "..."
}

Query: "What security challenges does Olis face and who is addressing them?"

Sub-queries:
1. "What are the security challenges in Olis?"
2. "Who is responsible for security in Olis?"
3. "What is the current security implementation status?"
4. "What security features are planned?"
5. "What security risks have been identified?"

git clone <repo-url>
cd olis-monorepo/apps/api-server

poetry install
# or
pip install -r requirements.txt

docker-compose up -d
# Starts: Milvus, Elasticsearch, Redis

cp .env.example .env
# Edit .env with your configuration

# Database
MILVUS_URI=http://localhost:19530
ES_URI=http://localhost:9200
REDIS_HOST=localhost
REDIS_PORT=6379

# LLM
RAG_LLM_PROVIDER=openai
OPENAI_API_KEY=sk-...
RAG_GENERATOR_MODEL=gpt-4o-mini

# Auth
JWT_SECRET=your-secret-key-change-in-production
JWT_ISSUER=olis-api

# Development with hot reload
uvicorn src.api_server.main:app --reload --port 8002

# Production
gunicorn src.api_server.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8002

curl http://localhost:8002/healthz
# Expected: {"status":"ok"}

# Run integration tests (requires running server)
RUN_INTEGRATION_TESTS=1 pytest tests/integration_tests/test_fastapi.py -v

import requests

def test_predict_endpoint():
    url = "http://localhost:8002/predict"
    headers = {"Authorization": f"Bearer {get_test_token()}"}
    data = {
        "query": "What are the Olis features?",
        "user_info": {"user_name": "test@olis.com"}
    }
    response = requests.post(url, json=data, headers=headers)
    assert response.status_code == 200
    assert "answer" in response.json()

pytest tests/unit_tests/ -v

# Using locust
locust -f tests/load_tests/locustfile.py --host=http://localhost:8002

# In main.py
logging.basicConfig(level=logging.DEBUG)

# Enable performance metrics
export PERF_LOG=true

# Server will emit:
# PERF {"event":"rag_start","ts":1738425600000,"request_id":"..."}
# PERF {"event":"retrieval_done","ts":1738425601000,"retrieval_ms":342,"docs_count":5}
# PERF {"event":"rag_end","ts":1738425603000,"total_ms":3124}

_debug_log(
    "main.py:lifespan",
    "retrieval_config",
    {"top_k": config.RETRIEVER_TOP_K, "use_reranker": config.USE_RERANKER},
    "H1"  # Hypothesis ID
)

{
  "event": "rag_end",
  "request_id": "uuid",
  "query_hash": "abc123def4",
  "retrieval_ms": 342,
  "context_ms": 12,
  "llm_ms": 2456,
  "post_ms": 23,
  "total_ms": 2833,
  "docs_count": 5,
  "context_chars": 4532,
  "model": "gpt-4o-mini",
  "num_predict": 500,
  "num_ctx": 4096
}

# Reduce top-K for faster retrieval
VECTOR_RETRIEVER_TOP_K=5
BM25_RETRIEVER_TOP_K=5
RETRIEVER_TOP_K=3

# Disable reranker if not needed
USE_RERANKER=false

# Skip LLM when no docs found
SKIP_LLM_ON_EMPTY_CONTEXT=true

# Limit doc/context characters
MAX_DOC_CHARS=1000
MAX_CONTEXT_CHARS=4000

# Faster but less comprehensive
USE_QUERY_DECOMPOSITION=false

# Avoid OpenAI API latency
RAG_LLM_PROVIDER=ollama
OLLAMA_MODEL=llama3.1
OLLAMA_NUM_PREDICT=300  # Faster generation

cd apps/api-server
docker build -t olis-api:latest .

docker run -d \
  -p 8002:8002 \
  -e MILVUS_URI=http://milvus:19530 \
  -e ES_URI=http://elasticsearch:9200 \
  -e REDIS_HOST=redis \
  -e OPENAI_API_KEY=sk-... \
  -e JWT_SECRET=production-secret \
  --name olis-api \
  olis-api:latest

version: '3.8'
services:
  olis-api:
    build: .
    ports:
      - "8002:8002"
    environment:
      MILVUS_URI: http://milvus:19530
      ES_URI: http://elasticsearch:9200
      REDIS_HOST: redis
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      JWT_SECRET: ${JWT_SECRET}
    depends_on:
      - milvus
      - elasticsearch
      - redis

  milvus:
    image: milvusdb/milvus:latest
    ports:
      - "19530:19530"
    volumes:
      - milvus_data:/var/lib/milvus

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
    ports:
      - "9200:9200"
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
    volumes:
      - es_data:/usr/share/elasticsearch/data

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

volumes:
  milvus_data:
  es_data:
  redis_data:

export JWT_SECRET=your-secret-key
# Restart server

# Verify Milvus is running
curl http://localhost:19530/healthz

# Check logs
docker logs milvus-standalone

# Restart Milvus
docker-compose restart milvus

# Verify connection
export MILVUS_URI=http://localhost:19530

# Check if documents exist
from pymilvus import connections, utility
connections.connect(uri="http://localhost:19530")
stats = utility.get_query_segment_info("docs_collection")
print(f"Docs in collection: {stats}")

export PERF_LOG=true
# Check which phase is slow:
# - retrieval_ms > 1000? Optimize Milvus/ES
# - llm_ms > 3000? Use smaller model or reduce context
# - context_ms > 500? Too many docs

# Reduce retrieval size
RETRIEVER_TOP_K=3

# Disable query decomposition
USE_QUERY_DECOMPOSITION=false

# Use faster LLM
RAG_GENERATOR_MODEL=gpt-4o-mini
OLLAMA_NUM_PREDICT=300

redis-cli -h localhost -p 6379 ping
# Expected: PONG

# Check Redis is running
docker ps | grep redis

# Verify connection params
echo $REDIS_HOST $REDIS_PORT

# Test connection
python -c "import redis; r=redis.Redis(host='localhost', port=6379); print(r.ping())"

@app.post("/predict")
async def predict(
    input: QueryInput,
    resources: AppResources = Depends(get_resources),
    current_user: Dict[str, Any] = Depends(get_current_user),
):
    # resources and current_user automatically injected
    pass

background_tasks.add_task(
    _update_memory_background,
    resources.session_memory,
    user_id,
    memory
)
# Response returns immediately, memory updated in background

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Initialize databases, LLM, etc.
    app.state.resources = AppResources(...)

    yield  # Server runs here

    # Shutdown: Close connections
    connections.disconnect("default")

AnswerWithKeywordsPrompt = """
You are Olis, an AI assistant that provides accurate answers based on provided context.

Context:
{context}

Question: {query}

Provide a JSON response with:
- callout_answer: Main answer (2-3 sentences)
- positive_list: Key points (list)
- negative_list: Caveats (list)
- info_list: Additional info (list)
- keywords: Relevant keywords (list)
"""

# Org A: docs_collection_orgaid, docs_bm25_orgaid
# Org B: docs_collection_orgbid, docs_bm25_orgbid

# Hard isolation at database level
org_res = resources.org_manager.get(org_id)
retriever = org_res["retriever"]  # Only sees org's data

from fastapi.responses import StreamingResponse

@app.post("/predict-stream")
async def predict_stream(input: QueryInput):
    async def generate():
        async for chunk in llm.astream(prompt):
            yield f"data: {chunk}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")