Building an OpenAI-Compatible Research Agent with Thinking Mode

# Final log showed:
logger.info(f"Tools: {result.get('tools_used')}")
# Output: Tools: []

# But the agent DID use tools:
# - search_web
# - mcp_fetch
# - geocode_location
# They just weren't tracked properly ❌

async def classify(self, query: str) -> tuple[Literal["simple", "medium", "complex"], int]:
    """
    Classify query complexity using LLM with keyword fallback.

    Returns:
        Tuple of (complexity_level, score_0_to_10)
    """
    # LLM scores 0-10
    score = await self._llm_classify(query)

    # Keyword safety net (e.g., "roadtrip", "unesco", "plan", "itinerary")
    keyword_level, keyword_score = self._keyword_fallback(query)

    # If keywords detect complex, override LLM
    if keyword_level == "complex":
        return keyword_level, keyword_score

    return self._score_to_level(score), score

# Reasoning (collapsed "Thinking..." section)
{
  "choices": [{
    "delta": {
      "role": "assistant",
      "reasoning_content": "🔍 Checking semantic cache...\n"
    }
  }]
}

# Final answer (visible response)
{
  "choices": [{
    "delta": {
      "role": "assistant",
      "content": "Het weer in Almere is op dit moment..."
    }
  }]
}

async def stream_response(agent, query: str, user_id: str, model: str):
    """Stream agent response in SSE format with reasoning support."""
    chat_id = f"chatcmpl-{uuid.uuid4()}"
    created = int(time.time())
    sent_thoughts = set()

    try:
        final_state = None

        # Stream events from LangGraph
        async for event in agent.run_streaming(query, user_id=user_id):
            if isinstance(event, dict):
                # Capture final state from "save" node
                if "save" in event:
                    final_state = event["save"]
                    continue

                # Stream thoughts as reasoning_content (collapsed section)
                for node_name, node_state in event.items():
                    if isinstance(node_state, dict) and "thoughts" in node_state:
                        new_thoughts = node_state.get("thoughts", [])
                        for thought in new_thoughts:
                            if thought not in sent_thoughts:
                                sent_thoughts.add(thought)

                                # Send as reasoning_content
                                data = {
                                    "id": chat_id,
                                    "object": "chat.completion.chunk",
                                    "created": created,
                                    "model": model,
                                    "choices": [{
                                        "index": 0,
                                        "delta": {
                                            "role": "assistant",
                                            "reasoning_content": f"{thought}\n"
                                        },
                                        "finish_reason": None
                                    }]
                                }
                                yield f"data: {json.dumps(data)}\n\n"

        # After loop, send final answer
        if final_state:
            final_response = final_state.get("final_response", "")

            # Stream final answer in chunks
            chunk_size = 50
            for i in range(0, len(final_response), chunk_size):
                chunk = final_response[i:i + chunk_size]
                data = {
                    "id": chat_id,
                    "object": "chat.completion.chunk",
                    "created": created,
                    "model": model,
                    "choices": [{
                        "index": 0,
                        "delta": {"role": "assistant", "content": chunk},
                        "finish_reason": None
                    }]
                }
                yield f"data: {json.dumps(data)}\n\n"

            # Send finish
            yield f"data: {json.dumps({...})}\n\n"
            yield "data: [DONE]\n\n"

    except Exception as e:
        logger.error(f"Streaming error: {e}")

prompt = f"""Answer the user's query based on the search results provided.

Query: {query}

Search Results:
{format_search_results(state["search_results"][:10])}

INSTRUCTIONS:
First, think through the information step-by-step inside <reasoning> tags.
Then, provide your final answer inside <answer> tags in the SAME LANGUAGE as the query.

Format:
<reasoning>
[Your step-by-step analysis of the search results, identifying key information,
resolving conflicts, etc.]
</reasoning>

<answer>
[Your final, well-structured answer with sources, in the query's language]
</answer>

Requirements for the answer:
- Cite sources using [Source: URL]
- Be comprehensive but concise
- Use markdown formatting
- If sources conflict, mention both perspectives
"""

response = await llm.ainvoke(prompt)
full_content = response.content

# Extract reasoning
if "<reasoning>" in full_content and "</reasoning>" in full_content:
    reasoning_start = full_content.find("<reasoning>") + len("<reasoning>")
    reasoning_end = full_content.find("</reasoning>")
    reasoning = full_content[reasoning_start:reasoning_end].strip()

# Extract answer
if "<answer>" in full_content and "</answer>" in full_content:
    answer_start = full_content.find("<answer>") + len("<answer>")
    answer_end = full_content.find("</answer>")
    answer = full_content[answer_start:answer_end].strip()

# Add reasoning to thoughts (appears in "Thinking..." section)
if reasoning:
    state["thoughts"] = [*state["thoughts"], f"🧠 LLM Reasoning:\n{reasoning}"]

# Set clean answer as final response
state["final_response"] = answer

# ❌ WRONG - Mutates in place (LangGraph misses the change)
state["tools_used"].append("search_web")
state["thoughts"].extend(["Step 1", "Step 2"])

# ✅ CORRECT - Creates new list (LangGraph detects the change)
state["tools_used"] = [*state["tools_used"], "search_web"]
state["thoughts"] = state["thoughts"] + ["Step 1", "Step 2"]

# Before fix:
logger.info(f"Tools: {result.get('tools_used')}")
# Output: Tools: []  ❌

# After fix:
logger.info(f"Tools: {result.get('tools_used')}")
# Output: Tools: ['search_web', 'mcp_fetch', 'geocode_location']  ✅

# agent/workflows/research.py
state["search_results"] = search_results.get("results", [])
state["tools_used"] = [*state["tools_used"], "searxng"]
state["thoughts"] = [*state["thoughts"], f"✓ Found {len(state['search_results'])} results"]

# agent/workflows/planning.py
state["artifacts"] = [*state["artifacts"], {
    "type": "pdf",
    "url": url,
    "filename": f"itinerary_{state['conversation_id']}.pdf"
}]
state["tools_used"] = [*state["tools_used"], "artifacts"]

# agent/core/agent.py (ReAct agent)
state["thoughts"] = [*state["thoughts"],
    f"✓ ReAct agent completed in {result.get('iterations', 0)} iterations",
    f"  Tools used: {', '.join(result['tools_used'])}"
]
state["tools_used"] = state["tools_used"] + result["tools_used"]

class MCPFetchTool:
    def _ensure_initialized(self):
        """Initialize MCP session with proper handshake."""
        if self._initialized:
            return

        # Step 1: Send initialize request
        init_result = self._rpc(
            "initialize",
            {
                "protocolVersion": "2024-11-05",
                "clientInfo": {"name": "research-agent", "version": "1.0.0"},
                "capabilities": {},
            },
        )
        logger.info("MCP initialized")

        # Step 2: Send notifications/initialized (CRITICAL!)
        self._notify("notifications/initialized")
        logger.info("MCP session ready")

        self._initialized = True

    def _rpc(self, method: str, params: dict) -> dict:
        """Make MCP JSON-RPC request with SSE response parsing."""
        self._id += 1
        payload = {"jsonrpc": "2.0", "id": self._id, "method": method}
        if params:
            payload["params"] = params

        with self.client.stream("POST", self.mcp_url, json=payload) as r:
            # Parse SSE format (data: prefix)
            for evt in _parse_sse_lines(r.iter_lines()):
                if evt["event"] == "message":
                    msg = json.loads(evt["data"])
                    if msg.get("id") == self._id:
                        return msg

    def fetch(self, url: str) -> dict:
        """Fetch web page content."""
        self._ensure_initialized()

        result = self._rpc("tools/call", {
            "name": "fetch",
            "arguments": {"url": url}
        })

        content_items = result["result"]["content"]
        text_content = ""
        for item in content_items:
            if item.get("type") == "text":
                text_content += item["text"]

        return {"url": url, "content": text_content, "status": 200}

def _parse_sse_lines(lines: Iterator[str]) -> Iterator[Dict[str, Any]]:
    """Parse Server-Sent Events format."""
    event_name = "message"
    data_buf: list[str] = []

    for line in lines:
        if line == "":
            if data_buf:
                yield {"event": event_name, "data": "\n".join(data_buf)}
            event_name = "message"
            data_buf = []
            continue

        m = EVENT_LINE_RE.match(line)  # event: ...
        if m:
            event_name = m.group(1).strip()
            continue

        m = DATA_LINE_RE.match(line)  # data: ...
        if m:
            data_buf.append(m.group(1))

async def classify(self, query: str) -> tuple[Literal["simple", "medium", "complex"], int]:
    # Step 1: Check keyword patterns first
    keyword_level, keyword_score = self._keyword_fallback(query)

    # If keywords say complex, trust that (LLM sometimes underestimates)
    if keyword_level == "complex":
        logger.info(f"Classified (keyword override: {keyword_score}) as 'complex'")
        return keyword_level, keyword_score

    # Step 2: Use LLM for nuanced classification
    score = await self._llm_classify(query)
    level = self._score_to_level(score)

    # Step 3: Sanity check
    if score == 0 and keyword_level in ["medium", "complex"]:
        logger.warning(f"LLM returned 0 but keywords suggest {keyword_level}")
        return keyword_level, keyword_score

    return level, score

def _keyword_fallback(self, query: str) -> tuple:
    """Keyword-based classification."""
    query_lower = query.lower()

    complex_keywords = [
        "plan", "itinerary", "roadtrip", "route", "journey",
        "create", "design", "develop", "build",
        "compare", "analyze", "research",
        "week", "month", "dagen", "daagse",
        "hotel", "accommodation", "restaurant", "budget",
        "unesco", "werelderfgoed"
    ]

    # Multiple complex indicators = very complex
    complex_matches = sum(1 for kw in complex_keywords if kw in query_lower)
    if complex_matches >= 2:
        return "complex", 9
    elif complex_matches == 1:
        return "complex", 8

    # Fallback to simple/medium based on length
    word_count = len(query.split())
    if word_count <= 5:
        return "simple", 3
    elif word_count <= 15:
        return "medium", 5
    else:
        return "complex", 7

prompt = f"""You are a query complexity analyzer. Rate this query's complexity from 0-10.

SCORING RULES:

0-3 = SIMPLE (one search, direct fact)
Examples:
- "What's the weather?" → 2
- "What time is it?" → 1

4-6 = MEDIUM (multiple searches, some analysis)
Examples:
- "Compare Python vs JavaScript" → 5
- "Explain how solar panels work" → 5

7-10 = COMPLEX (multi-step planning, requires multiple tools and reasoning)
Examples:
- "Plan a 5-day roadtrip with hotels and restaurants" → 9
- "Research and compare 5 laptops with pros/cons" → 8

QUERY TO RATE: "{query}"

IMPORTANT:
- If query mentions "plan", "roadtrip", "itinerary", "week", "days" → score 7-10
- If query asks for routes, hotels, AND restaurants → score 8-10
- Respond with ONLY ONE NUMBER (0-10)

YOUR SCORE:"""

class SupabaseCheckpointer:
    """Supabase-based checkpointer for LangGraph."""

    def __init__(self, client):
        self.client = client

    async def aput(
        self,
        config: Dict[str, Any],
        checkpoint: Checkpoint,
        metadata: Dict[str, Any],
        new_versions: Dict[str, Any],
    ) -> Dict[str, Any]:
        """Async version - wraps sync put."""
        await asyncio.to_thread(self.put, config, checkpoint, metadata)
        return config

    def put(
        self,
        config: Dict[str, Any],
        checkpoint: Checkpoint,
        metadata: Dict[str, Any]
    ) -> None:
        """Save checkpoint to Supabase."""
        thread_id = config["configurable"]["thread_id"]

        checkpoint_data = {
            "thread_id": thread_id,
            "checkpoint": checkpoint,
            "metadata": metadata,
            "created_at": datetime.now().isoformat()
        }

        self.client.table("checkpoints").upsert(
            checkpoint_data,
            on_conflict="thread_id"
        ).execute()

    async def aget_tuple(self, config: Dict[str, Any]):
        """Async version - wraps sync get_tuple."""
        return await asyncio.to_thread(self.get_tuple, config)

    def get_tuple(self, config: Dict[str, Any]):
        """Load checkpoint from Supabase."""
        thread_id = config["configurable"]["thread_id"]

        result = self.client.table("checkpoints")\
            .select("*")\
            .eq("thread_id", thread_id)\
            .order("created_at", desc=True)\
            .limit(1)\
            .execute()

        if result.data:
            data = result.data[0]
            return CheckpointTuple(
                config=config,
                checkpoint=data["checkpoint"],
                metadata=data.get("metadata", {}),
                parent_config=None
            )
        return None

# Het weer in Almere

## Actuele situatie

Het weer in Almere is op dit moment wisselend en koud:

- **Buienradar** meldt dat er op dit moment een gebied met **(natte) sneeuw**
  over Almere trekt, met een temperatuur van **-0,5°C** (onder het vriespunt)
  en een matige zuidelijke wind
  [Source: https://www.buienradar.nl/weer/almere_stad/nl/2759879]

- **Weerplaza** rapporteert daarentegen dat het vanavond **bewolkt maar zo
  goed als droog** blijft, met afkoeling naar **2°C** vannacht en zwakke wind
  [Source: https://www.weerplaza.nl/nederland/almere-stad/5445/]

## Verwachting komende dagen

- **Vannacht**: Afkoeling tot ongeveer 2°C met zwakke wind
- **Morgen**: Droog en zonnig, maar koud met een gevoelstemperatuur onder
  het vriespunt [Source: https://www.weeronline.nl/Europa/Nederland/Almere/4058927]

## Advies

Houd er rekening mee dat het vrij koud is (tussen -0,5°C en 2°C) en er lokaal
**gladheid** kan ontstaan door bevriezing, vooral als de natte sneeuw van
Buienradar actief is in de regio.

# 7-Daagse UNESCO Roadtrip door België

## 🗺️ Route Overzicht

**Totale afstand:** 487 km
**Budget:** €1000 (€142/dag)
**UNESCO Sites:** 5 locaties

## 📅 Dag-per-Dag Planning

### Dag 1: Brugge - Historisch Centrum
- **UNESCO Site**: Historisch centrum van Brugge
- **Hotel**: Budget Hotel Brugge (€65/nacht)
- **Restaurant**: De Stove (€25)
- **Bezienswaardigheden**: Belfort, Marktplein, kanalen

[... Full 7-day itinerary ...]

## 💰 Budget Breakdown

| Categorie | Kosten |
|-----------|--------|
| Hotels (6 nachten) | €420 |
| Restaurants | €350 |
| Brandstof | €80 |
| Entrees UNESCO | €120 |
| Reserve | €30 |
| **Totaal** | **€1000** |

## 📍 Interactieve Kaart

[Link to generated route map]

## 🎯 Tips

- Boek hotels minimaal 2 weken van tevoren voor deze prijzen
- Parkeren bij UNESCO sites kan uitdagend zijn - gebruik P+R
- Veel UNESCO sites bieden combinatietickets

# LLM Configuration
OLLAMA_LLM_URL=https://ollama.ictq.xyz
OLLAMA_LLM_MODEL=kimi-k2.5:cloud

# Embeddings (local instance)
OLLAMA_EMBEDDING_URL=http://10.106.0.49:11434
OLLAMA_EMBEDDING_MODEL=nomic-embed-text
OLLAMA_EMBEDDING_DIMENSION=768

# MCP Servers (via Toolhive proxy)
MCP_FETCH_URL=http://mcp-fetch-proxy.toolhive-system:8080/mcp
MCP_TIME_URL=http://mcp-time-proxy.toolhive-system:8080/mcp

# Search & Geocoding
SEARXNG_URL=http://searxng.search-system:8080
NOMINATIM_URL=https://nominatim.openstreetmap.org
OSRM_URL=http://router.project-osrm.org

# Supabase
SUPABASE_URL=https://your-project.supabase.co
SUPABASE_SERVICE_KEY=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

# Redis Cache
REDIS_URL=redis://redis.cache-system:6379
REDIS_PASSWORD=your-password
CACHE_TTL_SEARCH=3600
CACHE_TTL_GEOCODE=86400

# User Configuration
USER_ID=sander

apiVersion: apps/v1
kind: Deployment
metadata:
  name: research-agent
  namespace: ai-agents
spec:
  replicas: 2
  selector:
    matchLabels:
      app: research-agent
  template:
    metadata:
      labels:
        app: research-agent
    spec:
      containers:
      - name: research-agent
        image: registry.ictq.xyz/research-agent:latest
        ports:
        - containerPort: 8000
        env:
        - name: OLLAMA_LLM_URL
          value: "https://ollama.ictq.xyz"
        - name: OLLAMA_LLM_MODEL
          value: "kimi-k2.5:cloud"
        envFrom:
        - secretRef:
            name: research-agent-secrets
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: research-agent
  namespace: ai-agents
spec:
  selector:
    app: research-agent
  ports:
  - protocol: TCP
    port: 8000
    targetPort: 8000
  type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: research-agent
  namespace: ai-agents
  annotations:
    cert-manager.io/cluster-issuer: letsencrypt-prod
    nginx.ingress.kubernetes.io/proxy-read-timeout: "300"
    nginx.ingress.kubernetes.io/proxy-send-timeout: "300"
spec:
  ingressClassName: nginx
  tls:
  - hosts:
    - research-agent.ictq.xyz
    secretName: research-agent-tls
  rules:
  - host: research-agent.ictq.xyz
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: research-agent
            port:
              number: 8000

# ❌ BAD
state["tools_used"].append("search")

# ✅ GOOD
state["tools_used"] = [*state["tools_used"], "search"]

<reasoning>
[Internal thinking - sent to reasoning_content]
</reasoning>

<answer>
[User-facing response - sent to content]
</answer>

# Cache search results (1 hour)
CACHE_TTL_SEARCH = 3600

# Cache geocoding (24 hours - locations don't change)
CACHE_TTL_GEOCODE = 86400

# Cache embeddings (indefinite - deterministic)
# Store in Supabase with query hash

# ❌ Sequential (slow)
result1 = await tool1.run()
result2 = await tool2.run()
result3 = await tool3.run()

# ✅ Parallel (3x faster)
results = await asyncio.gather(
    tool1.run(),
    tool2.run(),
    tool3.run()
)

# Too small = more overhead
chunk_size = 10  # ❌

# Too large = delayed feedback
chunk_size = 500  # ❌

# Just right = smooth streaming
chunk_size = 50  # ✅

# Simple queries: Fast model
if complexity <= 3:
    model = "qwen2.5:32b"  # Fast, cheap

# Complex queries: Powerful model
else:
    model = "kimi-k2.5:cloud"  # Slow, expensive, high quality

# Query metrics
- complexity_distribution: Histogram of scores 0-10
- query_duration_by_complexity: P50, P95, P99
- cache_hit_rate: Percentage of cache hits
- tool_usage_frequency: Count per tool

# Error metrics
- llm_errors: Count + error types
- tool_errors: Count per tool
- timeout_rate: Queries exceeding thresholds

# User metrics
- queries_per_user: Distribution
- preferred_language: Language breakdown
- satisfaction_score: If feedback enabled

# Structured logging with context
logger.info(
    "Query completed",
    extra={
        "query_id": query_id,
        "user_id": user_id,
        "complexity": complexity,
        "duration_ms": duration_ms,
        "tools_used": tools_used,
        "cache_hit": cache_hit,
    }
)

# OpenTelemetry integration
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("research_query") as span:
    span.set_attribute("query.complexity", complexity)
    span.set_attribute("query.language", language)

    with tracer.start_as_current_span("search"):
        results = await search_tool.search(query)

    with tracer.start_as_current_span("llm_synthesis"):
        answer = await llm.ainvoke(prompt)