We are going to build a long-context research agent that ingests technical documents up to 131K tokens and returns structured JSON analysis. It is aimed at engineers who need to extract insights from large bodies of text without unpredictable token costs. Because Oxlo.ai uses request-based pricing, we can send entire whitepapers in a single call and the cost does not scale with input length.
What you'll need
- An Oxlo.ai API key
- Python 3.10 or newer
- The OpenAI SDK installed:
pip install openai
Step 1: Verify connectivity with a lightweight model
Before we optimize anything, I confirm the client is configured correctly. I always start with a single ping to a fast model to rule out network or credential issues.
from openai import OpenAI
import os
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY"),
)
ping = client.chat.completions.create(
model="llama-3.3-70b",
messages=[{"role": "user", "content": "Reply with exactly: connection OK"}],
max_tokens=10,
)
print(ping.choices[0].message.content)
Step 2: Design the system prompt
A vague prompt is the enemy of latency and consistency. I keep the system prompt constrained, assign a persona, and forbid markdown wrappers so JSON mode works cleanly.
SYSTEM_PROMPT = """You are a high-performance research parser. Read the technical document and emit a structured analysis.
Rules:
1. Output strictly valid JSON. No markdown code fences, no commentary.
2. Top-level keys must be: summary, key_entities, risks, action_items.
3. summary must be under 40 words.
4. key_entities, risks, and action_items must be arrays of strings.
5. If the document contains no technical content, return empty arrays and a null summary."""
Step 3: Use JSON mode to eliminate parsing overhead
Parsing free text with regex is fragile and slow. Oxlo.ai supports OpenAI-compatible JSON mode, so we can request valid JSON straight from the model and load it with the standard library.
import json
def extract_structured(document: str) -> dict:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": document},
],
response_format={"type": "json_object"},
max_tokens=1024,
)
return json.loads(response.choices[0].message.content)
sample = "The migration uses blue-green deployments. Risk: database schema drift."
print(extract_structured(sample))
Step 4: Stream tokens to improve perceived latency
Time-to-first-token matters more than total generation time for interactive tools. I enable streaming for the summary layer so users see progress immediately.
def stream_summary(document: str):
stream = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": "Summarize the input in one sentence. Plain text only."},
{"role": "user", "content": document},
],
stream=True,
max_tokens=80,
)
for chunk in stream:
token = chunk.choices[0].delta.content
if token:
print(token, end="", flush=True)
print()
stream_summary("Kubernetes operators automate complex stateful application management through custom resources.")
Step 5: Route complex reasoning to the right model
Not every subtask needs a 70B+ parameter model. I use a fast classifier to decide whether to send the full context to a lightweight model or to a heavy reasoning model.
def route_and_answer(query: str, context: str) -> str:
router = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": "Classify the query as 'simple' or 'complex'. Reply with one word only."},
{"role": "user", "content": f"Query: {query}"},
],
max_tokens=10,
)
label = router.choices[0].message.content.strip().lower()
target = "kimi-k2.6" if label == "complex" else "llama-3.3-70b"
response = client.chat.completions.create(
model=target,
messages=[
{"role": "system", "content": "Answer the question using the provided context. Be concise."},
{"role": "user", "content": f"Context: {context}\n\nQuestion: {query}"},
],
max_tokens=512,
)
return response.choices[0].message.content
print(route_and_answer("What is the failure mode?", "The system uses a single leader with no failover."))
Step 6: Ground output with function calling
Hallucinated entity names break downstream pipelines. I expose a search tool so the model can verify terms before it returns them, which reduces error rates.
TOOLS = [
{
"type": "function",
"function": {
"name": "search_web",
"description": "Search the web for a technical term.",
"parameters": {
"type": "object",
"properties": {
"term": {"type": "string"}
},
"required": ["term"],
},
},
}
]
def grounded_extraction(document: str):
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": document},
],
tools=TOOLS,
tool_choice="auto",
)
msg = response.choices[0].message
return {
"structured_json": msg.content,
"requested_tools": [tc.function.name for tc in (msg.tool_calls or [])],
}
print(grounded_extraction("Explain CRDTs and their use in distributed databases."))
Step 7: Assemble the long-context pipeline
Now we wire everything into a single agent class. Because Oxlo.ai pricing is per request, we can pass a large architecture RFC to kimi-k2.6 in one shot without input length driving up cost.
class ResearchAgent:
def __init__(self, client):
self.client = client
def analyze(self, document: str, stream: bool = False) -> dict:
extraction = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": document},
],
response_format={"type": "json_object"},
)
report = json.loads(extraction.choices[0].message.content)
if stream:
print("Live summary: ", end="", flush=True)
for chunk in self.client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": "Write a one-sentence plain-text summary."},
{"role": "user", "content": document},
],
stream=True,
max_tokens=100,
):
token = chunk.choices[0].delta.content
if token:
print(token, end="", flush=True)
print()
return report
agent = ResearchAgent(client)
Run it
Here is how I test the finished agent against a realistic system design document.
if __name__ == "__main__":
design_doc = """
System Design: Distributed Task Queue
The queue is built on Redis Streams with consumer groups for horizontal scaling.
Producers push JSON payloads to a stream key. Consumers read in batches of 100.
Risk: during a partition rebalance, in-flight messages can be double-processed.
Mitigation: consumers must be idempotent. A dead-letter stream captures failures
after three retries. Monitoring is handled by Prometheus exporters on each node.
"""
result = agent.analyze(design_doc, stream=True)
print("\n--- Final Report ---")
print(json.dumps(result, indent=2))
Example output:
Live summary: A Redis Streams task queue uses consumer groups and idempotent consumers to handle at-least-once delivery.
--- Final Report ---
{
"summary": "Redis Streams task queue with consumer groups and at-least-once delivery.",
"key_entities": [
"Redis Streams",
"Consumer Groups",
"Prometheus Exporters",
"Dead-Letter Stream"
],
"risks": [
"Double-processing during partition rebalance"
],
"action_items": [
"Implement idempotent consumers",
"Configure dead-letter stream after three retries"
]
}
Next steps
Two concrete directions to take this agent further.
First, add a retrieval step. Use Oxlo.ai's embeddings endpoint with bge-large to chunk and index your document corpus, then inject only the most relevant sections into the long-context window. This keeps requests focused and fast.
Second, wrap the agent in a FastAPI service and run it under load. Because Oxlo.ai charges per request rather than per token, your cost stays flat even when users submit massive documents. You can see the details on the pricing page.
Top comments (0)