""" TokioAI Engine v2.0 -- Main Pipeline Orchestrates: Router -> Memory -> Guided Generation -> Verifier -> Output The pipeline wraps any Ollama-compatible model and transforms it into an elite AI agent with multi-step tool calling. Key change from v1: when force_intent is None, the model freely chooses between tool calls and text responses. This enables investigation chains. """ import json import sys import os import re import time import requests from typing import Any, Callable, Dict, List, Optional, Tuple from .router import classify_intent, build_system_prompt, IntentType from .memory import EngineMemory from .guided import build_ollama_request, parse_guided_output from .verifier import Verifier, RetryStrategy, VerificationResult from .schemas import TOOL_DEFINITIONS, tools_to_prompt_text, validate_tool_call class EngineConfig: """Pipeline configuration.""" def __init__( self, ollama_host: str = None, model_name: str = "tokioai-coder", max_retries: int = 2, temperature: float = 0.3, max_tokens: int = 4096, use_guided_gen: bool = True, use_memory: bool = True, memory_path: str = None, verbose: bool = False, ): self.ollama_host = ollama_host or os.getenv( "TOKIOAI_ENGINE_HOST", os.getenv("OLLAMA_HOST", "http://localhost:11434") ) self.model_name = model_name self.max_retries = max_retries self.temperature = temperature self.max_tokens = max_tokens self.use_guided_gen = use_guided_gen self.use_memory = use_memory self.memory_path = memory_path self.verbose = verbose class PipelineResult: """Result from the pipeline.""" def __init__( self, output_type: str, # "tool_call", "text" tool_calls: List[Dict] = None, text: str = "", intent: str = "", confidence: float = 0.0, attempts: int = 1, latency_ms: int = 0, from_memory: bool = False, ): self.output_type = output_type self.tool_calls = tool_calls or [] self.text = text self.intent = intent self.confidence = confidence self.attempts = attempts self.latency_ms = latency_ms self.from_memory = from_memory def to_openai_message(self) -> Dict: """Convert to OpenAI-style assistant message.""" if self.output_type == "tool_call" and self.tool_calls: return { "role": "assistant", "content": None, "tool_calls": [ { "id": f"call_{i}", "type": "function", "function": { "name": tc["name"], "arguments": json.dumps(tc["arguments"]), }, } for i, tc in enumerate(self.tool_calls) ], } return { "role": "assistant", "content": self.text, } def __repr__(self): if self.output_type == "tool_call": names = [tc["name"] for tc in self.tool_calls] return f"PipelineResult(tool_calls={names}, attempts={self.attempts}, {self.latency_ms}ms)" return f"PipelineResult(text={self.text[:80]!r}, attempts={self.attempts}, {self.latency_ms}ms)" class Pipeline: """ The TokioAI Engine pipeline. Usage: engine = Pipeline(EngineConfig(model_name="tokioai-coder")) result = engine.process("check disk space") # result.tool_calls = [{"name": "execute_local", "arguments": {"command": "df -h"}}] """ def __init__(self, config: EngineConfig = None): self.config = config or EngineConfig() self.verifier = Verifier() self.retry_strategy = RetryStrategy(max_retries=self.config.max_retries) self.memory = EngineMemory(self.config.memory_path) if self.config.use_memory else None self.conversation_history: List[Dict] = [] self._stats = { "total_requests": 0, "tool_calls": 0, "text_responses": 0, "retries": 0, "failures": 0, "total_latency_ms": 0, } def process( self, user_message: str, conversation_history: List[Dict] = None, force_intent: str = None, ) -> PipelineResult: """ Process a user message through the full pipeline. Args: user_message: The user's input conversation_history: Previous messages (optional) force_intent: Override intent classification - "text": force text-only response (no tool calls allowed) - "tool_call": force tool call - None: model decides freely (default for investigation rounds) Returns: PipelineResult with tool calls or text response """ start_time = time.time() self._stats["total_requests"] += 1 history = conversation_history or self.conversation_history # === STAGE 1: ROUTE === if force_intent: intent, confidence = force_intent, 1.0 else: intent, confidence = classify_intent(user_message, history) if self.config.verbose: print(f"[ENGINE] Intent: {intent} (confidence: {confidence:.2f})") # === PURE TEXT SHORT-CIRCUIT === # For greetings and thanks, strip conversation history to prevent # context bleeding (e.g. "hola" after HA investigation should NOT talk about HA). # BUT NOT for affirmations ("dale", "si", "ok") which need context. import re as _re _GREETING_PATS = [ r'^(hi|hello|hey|hola|que\s+tal|buenos?\s+d[ií]as?|buenas?\s+(?:tardes?|noches?))\s*[.!?]*$', r'^(thanks?|thank\s+you|gracias|thx|de\s+nada)\s*[.!?]*$', r'^(como\s+(va|estas?|andas?|te\s+va)|how\s+are\s+you)\s*[.!?]*$', r'^(todo\s+bien|bien\s+y\s+tu)\s*[.!?]*$', r'\b(who\s+are\s+you|what\s+can\s+you|tell\s+me\s+about\s+yourself)\b', ] _msg_lower = user_message.lower().strip() is_pure_text = ( intent == "text" and confidence >= 0.9 and any(_re.match(p, _msg_lower, _re.IGNORECASE) for p in _GREETING_PATS) ) # === STAGE 2: MEMORY LOOKUP === few_shot = [] if self.memory and self.config.use_memory and not is_pure_text: few_shot = self.memory.retrieve(user_message, intent, top_k=3) if self.config.verbose and few_shot: print(f"[ENGINE] Memory: {len(few_shot)} relevant examples found") # === STAGE 3: BUILD PROMPT === system_prompt = build_system_prompt( intent=intent, few_shot_examples=few_shot, ) # Build messages messages = [{"role": "system", "content": system_prompt}] # Add conversation history -- but for pure text (greetings etc), # only include the LAST 2 messages to avoid context bleeding if is_pure_text: # For greetings/thanks: minimal or no history # This prevents "hola" from generating an HA status report # because the previous conversation was about HA if self.config.verbose: print(f"[ENGINE] Pure text detected, stripping history ({len(history)} msgs)") else: for msg in history[-20:]: messages.append(msg) # Add current user message messages.append({"role": "user", "content": user_message}) # === STAGE 4: GENERATE === # Use guided generation ONLY for definite tool_call intent on first request # For investigation rounds (force_intent=None with history), let model choose use_guided = self.config.use_guided_gen and intent == "tool_call" result = self._generate_with_retries( messages=messages, intent=intent, use_guided=use_guided, force_text_only=(force_intent == "text"), ) # === STAGE 5: RECORD === latency_ms = int((time.time() - start_time) * 1000) result.intent = intent result.confidence = confidence result.latency_ms = latency_ms self._stats["total_latency_ms"] += latency_ms if result.output_type == "tool_call": self._stats["tool_calls"] += 1 else: self._stats["text_responses"] += 1 # Save to memory (only successful results) if self.memory and result.output_type in ("tool_call", "text"): tool_name = result.tool_calls[0]["name"] if result.tool_calls else None output_text = ( json.dumps(result.tool_calls[0]) if result.tool_calls else result.text[:500] ) self.memory.add( user_input=user_message, model_output=output_text, intent=intent, success=True, tool_name=tool_name, latency_ms=latency_ms, ) # Update conversation history self.conversation_history.append({"role": "user", "content": user_message}) self.conversation_history.append(result.to_openai_message()) return result def add_tool_result(self, tool_name: str, result: str): """Add a tool execution result to the conversation.""" self.conversation_history.append({ "role": "tool", "content": result, "name": tool_name, }) def _generate_with_retries( self, messages: List[Dict], intent: str, use_guided: bool, force_text_only: bool = False, ) -> PipelineResult: """ Generate model output with retry logic on verification failure. Args: messages: Full message chain intent: Classified intent use_guided: Whether to use Ollama guided generation (JSON schema) force_text_only: If True, NEVER return tool calls (depth limit reached) """ for attempt in range(self.config.max_retries + 1): # Build request if use_guided: request = build_ollama_request( model=self.config.model_name, messages=messages, intent=intent, force_json=True, temperature=self.config.temperature, max_tokens=self.config.max_tokens, ) else: # Free-form generation: model can output tool calls OR text request = { "model": self.config.model_name, "messages": messages, "stream": False, "options": { "temperature": self.config.temperature, "num_predict": self.config.max_tokens, }, } # Include tools so Ollama can use native tool calling # UNLESS we're forcing text-only mode if not force_text_only and intent != "text": request["tools"] = TOOL_DEFINITIONS # Call Ollama try: raw_output, native_tool_calls = self._call_ollama(request) except Exception as e: if self.config.verbose: print(f"[ENGINE] Ollama error: {e}") self._stats["failures"] += 1 return PipelineResult( output_type="text", text=f"Error calling model: {e}", attempts=attempt + 1, ) if self.config.verbose: print(f"[ENGINE] Raw output (attempt {attempt+1}): {raw_output[:200]}") if native_tool_calls: print(f"[ENGINE] Native tool calls: {native_tool_calls}") # === NATIVE TOOL CALLS (Ollama returned tool_calls in response) === if native_tool_calls and not force_text_only: valid_calls = [] for tc in native_tool_calls: is_valid, err = validate_tool_call(tc["name"], tc["arguments"]) if is_valid: valid_calls.append(tc) elif self.config.verbose: print(f"[ENGINE] Invalid native tool call: {err}") if valid_calls: return PipelineResult( output_type="tool_call", tool_calls=valid_calls, attempts=attempt + 1, ) # === GUIDED OUTPUT (JSON constrained) === if use_guided and intent in ("tool_call", "decision"): parsed = parse_guided_output(raw_output, intent) if parsed["type"] == "tool_call" and not force_text_only: name = parsed.get("name", "") args = parsed.get("arguments", {}) is_valid, err = validate_tool_call(name, args) if is_valid: return PipelineResult( output_type="tool_call", tool_calls=[{"name": name, "arguments": args}], attempts=attempt + 1, ) elif self.config.verbose: print(f"[ENGINE] Guided validation failed: {err}") elif parsed["type"] == "text": return PipelineResult( output_type="text", text=parsed.get("response", raw_output), attempts=attempt + 1, ) # === FREE-FORM VERIFICATION === verification = self.verifier.verify(raw_output, expected_intent=intent) if verification.valid: if verification.output_type == "tool_call" and not force_text_only: # Model wants to call a tool -- allow it return PipelineResult( output_type="tool_call", tool_calls=verification.tool_calls, attempts=attempt + 1, ) else: # Text response (or forced text-only) text = verification.text or raw_output # Clean up: if model tried to output tool call JSON but we're # in text-only mode, don't show the raw JSON if force_text_only and verification.output_type == "tool_call": text = ( "Based on my investigation, I was unable to gather " "additional information. Here is what I found so far." ) return PipelineResult( output_type="text", text=text, attempts=attempt + 1, ) # Failed verification -- retry? if self.retry_strategy.should_retry(attempt, verification): self._stats["retries"] += 1 retry_msg = self.retry_strategy.get_retry_message(attempt, verification) messages.append({"role": "assistant", "content": raw_output}) messages.append({"role": "user", "content": retry_msg}) if self.config.verbose: print(f"[ENGINE] Retrying (attempt {attempt+2})...") continue else: # Give up -- return whatever we got self._stats["failures"] += 1 return PipelineResult( output_type="text", text=verification.text or raw_output, attempts=attempt + 1, ) # Should never reach here return PipelineResult(output_type="text", text="Engine error: max retries exceeded") def _call_ollama(self, request: Dict): """ Call Ollama API and return (text_output, native_tool_calls). Returns: Tuple of (str, list): - text content from the response - list of native tool calls (if any), each as {name, arguments} """ url = f"{self.config.ollama_host}/api/chat" try: resp = requests.post(url, json=request, timeout=120) if self.config.verbose and resp.status_code != 200: print(f"[ENGINE] Ollama HTTP {resp.status_code}: {resp.text[:500]}") resp.raise_for_status() data = resp.json() except requests.exceptions.ConnectionError: raise RuntimeError(f"Cannot connect to Ollama at {self.config.ollama_host}") except requests.exceptions.Timeout: raise RuntimeError("Ollama request timed out (120s)") except Exception as e: raise RuntimeError(f"Ollama request failed: {e}") message = data.get("message", {}) content = message.get("content", "") # Check for native tool calls from Ollama native_tool_calls = [] tool_calls_raw = message.get("tool_calls", []) if tool_calls_raw: for tc in tool_calls_raw: func = tc.get("function", {}) native_tool_calls.append({ "name": func.get("name", ""), "arguments": func.get("arguments", {}), }) return content, native_tool_calls def get_stats(self) -> Dict: """Return pipeline statistics.""" total = self._stats["total_requests"] or 1 return { "total_requests": self._stats["total_requests"], "tool_calls": self._stats["tool_calls"], "text_responses": self._stats["text_responses"], "retries": self._stats["retries"], "failures": self._stats["failures"], "avg_latency_ms": self._stats["total_latency_ms"] // total, "tool_call_rate": f"{100 * self._stats['tool_calls'] // total}%", } def reset(self): """Reset conversation history.""" self.conversation_history = []