TopolearnA notebook prototype and a reliable service differ in data freshness, concurrency, cost control and safety.
Moving from notebook to service raises a predictable set of questions.
Data freshness. Run ingestion as a scheduled or event-driven job, separate from the query service. Use stable document ids, an IngestionPipeline with a docstore, and a persistent vector database, so updates replace old chunks instead of duplicating them. Record which embedding model built each collection, and plan re-indexing when you change it.
Concurrency and latency. Use async methods (aquery, aretrieve, achat, agent.run) in async web servers so one slow LLM call doesn't block other users. Stream answers to the interface. Keep similarity_top_k and prompt size modest, and cache answers to frequent questions.
Cost. Most spending is LLM tokens: retrieved context times number of calls. Use the cheapest model that passes your evaluations, small response modes, and fewer agent steps. Track tokens per request through instrumentation.
Security.
- Enforce document permissions with metadata filters on every query, never with prompts.
- Treat retrieved text as untrusted: a document can contain prompt injection, text that tries to instruct the model. Give agents only the tools they need, and require confirmation for actions with side effects.
- Avoid code-executing agents (
CodeActAgent) outside a sandbox.
Serving workflows. Agents and workflows are ordinary Python, so you can wrap them in your own API. LlamaIndex also offers llamactl and LlamaAgents for running workflows as services and deploying them to its cloud; check the current docs for details.
Keep evaluating. Re-run your test set on every change to data processing, models or prompts, and sample real traffic for review.
Example
import asyncio
from llama_index.core import VectorStoreIndex
from llama_index.core.vector_stores import FilterOperator, MetadataFilter, MetadataFilters
async def answer(index: VectorStoreIndex, question: str, user_groups: list[str]) -> dict:
# Permission filter on EVERY query: only chunks tagged with the user's groups.
filters = MetadataFilters(filters=[
MetadataFilter(key="access_group", value=user_groups, operator=FilterOperator.IN)])
qe = index.as_query_engine(similarity_top_k=4, filters=filters)
response = await qe.aquery(question) # async: doesn't block other users
return {
"answer": str(response),
"sources": [n.node.metadata.get("file_name") for n in response.source_nodes],
}
async def main(index):
results = await asyncio.gather( # many users at once
answer(index, "What is the travel policy?", ["staff"]),
answer(index, "Show me salary bands", ["staff"]), # hr-only docs never retrieved
)
print(results)Remember
- Separate, incremental ingestion jobs with stable ids and a persistent store
- Async APIs, streaming and modest context for latency and cost
- Permissions via filters; treat retrieved text as untrusted
Watch out for
- Running ingestion inside the request path, so the first user after a change waits minutes and concurrent requests re-embed the same files.