What is the Cold Start Problem in AI Agents?
The cold start problem in AI agents refers to the situation where an agent begins operating with zero prior context, no accumulated knowledge, and no established behavioral patterns. Much like a new employee showing up on their first day without any onboarding materials, an AI agent experiencing a cold start lacks the foundational information needed to perform effectively from the outset.
In technical terms, cold start manifests when an agent is initialized without:
- Prior conversation history — no memory of past user interactions or decisions
- Pre-computed embeddings or indexes — vector stores and retrieval systems start empty
- Learned user preferences — no model of what the user likes, dislikes, or expects
- Environment calibration — no understanding of the operational context, API rate limits, or system constraints
- Baseline performance metrics — no historical data to guide confidence estimation or fallback behavior
This problem is not merely about empty databases. It encompasses the entire spectrum of statefulness in agent systems — from the low-level caching of LLM prompt prefixes to the high-level semantic understanding of a user's goals. Every new session, every fresh deployment, and every newly instantiated agent faces some degree of cold start unless explicitly mitigated.
Why It Matters
The consequences of cold start problems ripple across every dimension of agent performance:
- Latency spikes: Without cached computations, the agent must perform expensive operations from scratch — embedding generation, full context construction, and exhaustive tool discovery — all of which add measurable seconds to response time
- Degraded response quality: An agent with no historical context produces generic, untailored responses that fail to account for the user's specific domain, terminology, or previous problem-solving trajectory
- Hallucination amplification: Cold agents lack the grounding signals that come from accumulated interaction data, making them more susceptible to fabricating facts or invoking incorrect tools
- Resource waste: Redundant computation across sessions burns compute budget on tasks that could have been precomputed once and reused indefinitely
- User trust erosion: Users quickly notice when an agent doesn't remember them. The experience feels broken, impersonal, and fundamentally unreliable — leading to abandonment
In production systems serving thousands of concurrent users, cold starts are not a theoretical edge case. They happen continuously as new users join, sessions expire, cache entries are evicted, and deployments roll over. Every cold start is a potential failure point that degrades the aggregate reliability metrics of the entire agent fleet.
Common Manifestations of Cold Start Problems
Before diving into solutions, it's essential to recognize where cold start problems typically surface in agent architectures:
- RAG pipelines: A freshly initialized vector database contains no documents, so retrieval returns empty or low-quality results until indexing completes
- Tool selection: The agent lacks usage statistics about which tools are most reliable for which tasks, causing suboptimal tool routing
- Planning modules: Without prior examples of successful plans for similar tasks, the planner generates naive or overly complex decomposition strategies
- Reflex / guardrail systems: Content filters and safety classifiers may be overly permissive or restrictive until they accumulate sufficient calibration data
- Personalization layers: User embedding vectors don't exist yet, so recommendation and adaptation logic falls back to population-level averages
Strategies to Fix Cold Start Problems
1. Bootstrapping with Pre-built Knowledge Bases
The most direct mitigation is to pre-populate the agent's knowledge infrastructure before it ever serves a real request. Rather than waiting for organic accumulation, you inject a curated corpus of domain-relevant documents, common Q&A pairs, and canonical examples during the deployment phase.
This approach transforms the cold start from an empty state to a warm-but-generic state. The agent may still lack user-specific knowledge, but it won't be completely blind to the domain.
# bootstrap_knowledge.py
from typing import List, Dict
import json
from vector_store import VectorStore
from embedder import EmbeddingModel
class KnowledgeBootstrapper:
"""Pre-populates a vector store with curated seed documents
to eliminate the empty-index cold start scenario."""
def __init__(self, vector_store: VectorStore, embedder: EmbeddingModel):
self.vector_store = vector_store
self.embedder = embedder
def bootstrap_from_manifest(self, manifest_path: str) -> int:
"""Load a JSON manifest of seed documents and index them all.
The manifest format:
[
{
"id": "doc_001",
"content": "Full document text here...",
"metadata": {
"category": "api_reference",
"tags": ["authentication", "oauth"],
"priority": "high"
}
},
...
]
"""
with open(manifest_path, 'r') as f:
documents = json.load(f)
indexed_count = 0
batch_size = 50
for i in range(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
texts = [doc["content"] for doc in batch]
metadatas = [doc["metadata"] for doc in batch]
ids = [doc["id"] for doc in batch]
# Generate embeddings in bulk
embeddings = self.embedder.embed_batch(texts)
# Upsert into vector store
self.vector_store.upsert(
ids=ids,
embeddings=embeddings,
metadatas=metadatas,
documents=texts
)
indexed_count += len(batch)
print(f"Bootstrapped {indexed_count} seed documents into vector store.")
return indexed_count
def bootstrap_from_faq(self, faq: List[Dict[str, str]]) -> None:
"""Index FAQ pairs as retrievable documents with question-centric metadata."""
for i, pair in enumerate(faq):
doc_id = f"faq_{i:05d}"
# Store both question and answer as retrievable content
content = f"Q: {pair['question']}\nA: {pair['answer']}"
self.vector_store.upsert(
ids=[doc_id],
embeddings=[self.embedder.embed(content)],
metadatas=[{"type": "faq", "question": pair["question"]}],
documents=[content]
)
2. Few-Shot Prompting and In-Context Learning
When the agent's external memory is cold, you can compensate by enriching the prompt itself with carefully constructed examples. Few-shot exemplars injected into the system prompt or retrieved dynamically provide the agent with behavioral templates that guide reasoning even in the absence of historical data.
This technique works because LLMs are adept at pattern matching within their context window. A well-chosen set of 3-5 examples can dramatically improve output consistency and reduce the need for external state.
# few_shot_bootstrapper.py
from typing import List, Dict, Any
class FewShotPromptBuilder:
"""Constructs prompts with embedded examples to mitigate cold start
in the agent's reasoning layer."""
def __init__(self):
# Curated exemplars organized by task type
self.exemplar_library = {
"code_review": [
{
"input": "Review this Python function for security issues:\n"
"def get_user(user_id):\n"
" query = f\"SELECT * FROM users WHERE id = {user_id}\"\n"
" return db.execute(query)",
"output": "CRITICAL: SQL injection vulnerability detected.\n"
"The user_id is directly interpolated into the query string.\n"
"Fix: Use parameterized queries:\n"
" query = \"SELECT * FROM users WHERE id = ?\"\n"
" return db.execute(query, (user_id,))"
},
{
"input": "Review this code for error handling:\n"
"def read_config(path):\n"
" with open(path) as f:\n"
" return json.load(f)",
"output": "ISSUE: No error handling for missing file or invalid JSON.\n"
"Fix: Add try/except blocks for FileNotFoundError and JSONDecodeError."
}
],
"data_analysis": [
{
"input": "Analyze this sales data: [1200, 1350, 1100, 1400, 1250]",
"output": "Mean: 1260, Median: 1250, Range: 300, "
"Trend: Slight upward trend with volatility in month 3."
}
]
}
def build_prompt(self, task_type: str, user_query: str,
num_examples: int = 3) -> str:
"""Construct a prompt with relevant few-shot examples baked in."""
exemplars = self.exemplar_library.get(task_type, [])
selected = exemplars[:num_examples]
prompt_parts = [
f"You are an expert AI agent specialized in {task_type}.",
"Here are examples of how to handle similar tasks:\n"
]
for i, ex in enumerate(selected, 1):
prompt_parts.append(f"Example {i}:")
prompt_parts.append(f"INPUT: {ex['input']}")
prompt_parts.append(f"OUTPUT: {ex['output']}")
prompt_parts.append("") # Blank line separator
prompt_parts.append("Now, process the following new request with the "
"same level of detail and structure:\n")
prompt_parts.append(f"INPUT: {user_query}")
prompt_parts.append("OUTPUT:")
return "\n".join(prompt_parts)
def add_dynamic_exemplar(self, task_type: str, example: Dict[str, str]) -> None:
"""Allow the system to grow its exemplar library over time."""
if task_type not in self.exemplar_library:
self.exemplar_library[task_type] = []
self.exemplar_library[task_type].append(example)
3. Warm-Up Caches and Embedding Precomputation
A significant portion of cold start latency comes from computing embeddings and constructing context on the fly. By precomputing and caching these artifacts, you can serve the first request almost as quickly as subsequent ones.
This strategy involves running a warm-up routine immediately after deployment — before traffic is routed to the new instance. The warm-up phase precomputes embeddings for known tool descriptions, pre-fills LRU caches with common query patterns, and establishes connection pools to downstream services.
# warm_up_cache.py
import asyncio
import hashlib
from typing import List, Optional
from collections import OrderedDict
class AgentWarmUpCache:
"""Pre-warms caches to avoid cold start latency on first requests."""
def __init__(self, max_cache_size: int = 1000):
self.embedding_cache = OrderedDict()
self.max_cache_size = max_cache_size
self.tool_description_embeddings = {}
self._warmed_up = False
async def warm_up(self, tool_registry: List[dict],
common_queries: List[str],
embedder) -> None:
"""Execute full warm-up routine before accepting traffic.
This should be called during the deployment's readiness probe phase
so that the pod isn't marked as ready until warming is complete.
"""
print("Starting agent warm-up sequence...")
# Phase 1: Pre-embed all tool descriptions
await self._warm_tool_embeddings(tool_registry, embedder)
# Phase 2: Pre-embed common query templates
await self._warm_query_embeddings(common_queries, embedder)
# Phase 3: Pre-establish connection pools (simulated)
await self._warm_connections()
self._warmed_up = True
print(f"Warm-up complete. Cache ready with "
f"{len(self.embedding_cache)} pre-computed entries.")
async def _warm_tool_embeddings(self, tools: List[dict], embedder) -> None:
"""Pre-compute embeddings for every tool's description."""
descriptions = []
tool_ids = []
for tool in tools:
desc = f"{tool['name']}: {tool['description']}"
if 'parameters' in tool:
desc += f" Parameters: {tool['parameters']}"
descriptions.append(desc)
tool_ids.append(tool['name'])
embeddings = await embedder.embed_batch_async(descriptions)
for tool_id, embedding in zip(tool_ids, embeddings):
self.tool_description_embeddings[tool_id] = embedding
cache_key = f"tool_embed:{tool_id}"
self.embedding_cache[cache_key] = embedding
print(f" Pre-embedded {len(tool_ids)} tool descriptions")
async def _warm_query_embeddings(self, queries: List[str], embedder) -> None:
"""Pre-compute embeddings for common query patterns."""
if not queries:
return
embeddings = await embedder.embed_batch_async(queries)
for query, embedding in zip(queries, embeddings):
cache_key = hashlib.md5(query.encode()).hexdigest()
self.embedding_cache[cache_key] = embedding
print(f" Pre-embedded {len(queries)} common query patterns")
async def _warm_connections(self) -> None:
"""Simulate connection pool warm-up (in production, this would
actually open connections to databases, APIs, etc.)."""
await asyncio.sleep(0.1) # Simulated connection establishment
print(" Connection pools warmed")
def get_cached_embedding(self, key: str) -> Optional[List[float]]:
"""Retrieve a cached embedding, promoting it to most-recently-used."""
if key in self.embedding_cache:
# LRU promotion: move to end
self.embedding_cache.move_to_end(key)
return self.embedding_cache[key]
return None
def is_ready(self) -> bool:
"""Check if warm-up has completed — used by health check endpoints."""
return self._warmed_up
4. Hybrid Retrieval-Augmented Generation (RAG)
A pure vector-based RAG system suffers cold start because new documents aren't indexed. A hybrid approach combines vector search with traditional keyword-based retrieval (like BM25) and structured metadata queries. When the vector index is sparse, the keyword and metadata layers provide fallback retrieval paths that work immediately.
# hybrid_retriever.py
from typing import List, Tuple
import re
from collections import Counter
class HybridRetriever:
"""Combines vector search, BM25 keyword search, and metadata filtering
to provide robust retrieval even when the vector index is cold."""
def __init__(self, vector_store, document_store):
self.vector_store = vector_store
self.document_store = document_store # Stores raw documents + metadata
self.bm25_index = {} # In-memory keyword index (pre-built at startup)
def build_keyword_index(self, documents: List[dict]) -> None:
"""Pre-build a BM25-style keyword index from seed documents.
This index is immediately available even before embeddings are computed."""
for doc in documents:
doc_id = doc["id"]
# Tokenize and count term frequencies
tokens = re.findall(r'\w+', doc["content"].lower())
term_freqs = Counter(tokens)
# Compute inverse document frequency approximations
self.bm25_index[doc_id] = {
"terms": term_freqs,
"length": len(tokens),
"metadata": doc.get("metadata", {})
}
# Compute IDF values across the corpus
doc_count = len(documents)
term_doc_counts = Counter()
for doc in documents:
unique_terms = set(re.findall(r'\w+', doc["content"].lower()))
for term in unique_terms:
term_doc_counts[term] += 1
self.idf_values = {
term: (doc_count - count + 0.5) / (count + 0.5) + 1.0
for term, count in term_doc_counts.items()
}
def retrieve(self, query: str, top_k: int = 5) -> List[dict]:
"""Hybrid retrieval combining multiple signals."""
results = []
# Path 1: Vector similarity (may be empty during cold start)
try:
vector_results = self.vector_store.similarity_search(query, top_k)
for r in vector_results:
results.append({**r, "source": "vector", "score": r.get("score", 0)})
except Exception:
# Vector store might be empty — that's okay, we have fallbacks
pass
# Path 2: BM25 keyword search (always available after index build)
keyword_results = self._bm25_search(query, top_k)
for r in keyword_results:
if not any(res.get("id") == r["id"] for res in results):
results.append({**r, "source": "bm25", "score": r["bm25_score"]})
# Path 3: Metadata exact match (for structured queries)
metadata_results = self._metadata_search(query, top_k)
for r in metadata_results:
if not any(res.get("id") == r["id"] for res in results):
results.append({**r, "source": "metadata", "score": 1.0})
# Sort by relevance and return top_k
results.sort(key=lambda x: x.get("score", 0), reverse=True)
return results[:top_k]
def _bm25_search(self, query: str, top_k: int) -> List[dict]:
"""BM25-style keyword retrieval."""
query_terms = re.findall(r'\w+', query.lower())
scored_docs = []
avg_doc_length = sum(
idx["length"] for idx in self.bm25_index.values()
) / max(1, len(self.bm25_index))
k1, b = 1.5, 0.75 # BM25 parameters
for doc_id, index_data in self.bm25_index.items():
score = 0.0
for term in query_terms:
if term in self.idf_values:
idf = self.idf_values[term]
tf = index_data["terms"].get(term, 0)
# BM25 scoring formula
numerator = tf * (k1 + 1)
denominator = tf + k1 * (1 - b + b * (index_data["length"] / avg_doc_length))
score += idf * (numerator / denominator) if denominator != 0 else 0
if score > 0:
scored_docs.append({
"id": doc_id,
"bm25_score": score,
"metadata": index_data["metadata"]
})
scored_docs.sort(key=lambda x: x["bm25_score"], reverse=True)
return scored_docs[:top_k]
def _metadata_search(self, query: str, top_k: int) -> List[dict]:
"""Search documents by metadata fields extracted from query."""
# Simple heuristic: look for category:field or tag:field patterns
metadata_matches = []
pattern = r'(category|tag):(\w+)'
matches = re.findall(pattern, query.lower())
if not matches:
return []
for doc_id, index_data in self.bm25_index.items():
metadata = index_data["metadata"]
doc_category = metadata.get("category", "").lower()
doc_tags = [t.lower() for t in metadata.get("tags", [])]
for field, value in matches:
if field == "category" and value == doc_category:
metadata_matches.append({"id": doc_id, "metadata": metadata})
elif field == "tag" and value in doc_tags:
metadata_matches.append({"id": doc_id, "metadata": metadata})
return metadata_matches[:top_k]
5. Progressive Agent Initialization
Rather than attempting to fully initialize an agent before it serves any traffic, progressive initialization accepts that cold start is inevitable and manages it gracefully. The agent starts in a minimal viable state and progressively enriches its capabilities as it accumulates interactions.
This strategy is particularly effective because it converts the cold start from a binary failure mode into a gradual improvement curve. The first few interactions may be slower or less accurate, but the system transparently improves with each request.
# progressive_agent.py
from enum import Enum
from datetime import datetime, timedelta
from typing import Dict, Any, Optional
class AgentTier(Enum):
COLD = "cold" # No context, generic responses
WARMING = "warming" # Partial context, improving responses
HOT = "hot" # Rich context, optimal responses
DEGRADED = "degraded" # Context expired, needs refresh
class ProgressiveAgent:
"""An agent that gracefully handles cold starts by operating in tiers
and transparently upgrading its capability level."""
def __init__(self, user_id: str):
self.user_id = user_id
self.tier = AgentTier.COLD
self.interaction_count = 0
self.context_window = []
self.preferences: Dict[str, Any] = {}
self.last_interaction = None
self.session_start = datetime.now()
# Thresholds for tier progression
self.WARMING_THRESHOLD = 3 # Interactions needed to exit COLD
self.HOT_THRESHOLD = 10 # Interactions needed to exit WARMING
self.EXPIRY_HOURS = 24 # Time before context degrades
async def process(self, user_input: str) -> Dict[str, Any]:
"""Process user input, adapting behavior based on current tier."""
self.interaction_count += 1
self.last_interaction = datetime.now()
# Check for context expiry
if self.tier == AgentTier.HOT:
if datetime.now() - self.session_start > timedelta(hours=self.EXPIRY_HOURS):
self.tier = AgentTier.DEGRADED
await self._rehydrate_context()
# Route processing based on tier
if self.tier == AgentTier.COLD:
response = await self._process_cold(user_input)
elif self.tier == AgentTier.WARMING:
response = await self._process_warming(user_input)
elif self.tier == AgentTier.HOT:
response = await self._process_hot(user_input)
else: # DEGRADED
response = await self._process_degraded(user_input)
# Update tier based on accumulated context
self._recalculate_tier()
# Store interaction in context window
self.context_window.append({
"input": user_input,
"response": response,
"timestamp": datetime.now().isoformat(),
"tier": self.tier.value
})
# Trim context window to prevent unbounded growth
if len(self.context_window) > 50:
self.context_window = self.context_window[-50:]
return {
"response": response,
"tier": self.tier.value,
"interaction_count": self.interaction_count
}
async def _process_cold(self, user_input: str) -> str:
"""Cold tier: Use explicit clarification and generic patterns.
Acknowledge limitations transparently."""
# In cold tier, we explicitly ask clarifying questions
# and avoid making assumptions
clarification = (
"I'm getting up to speed on your preferences. "
"To give you the best response, could you clarify "
"a few things about your goals?"
)
# Still attempt to help with the actual query using generic reasoning
generic_response = await self._generic_reasoning(user_input)
return f"{clarification}\n\nBased on what I can see so far: {generic_response}"
async def _process_warming(self, user_input: str) -> str:
"""Warming tier: Use partial context with confidence flags."""
recent_context = self.context_window[-3:] if self.context_window else []
context_summary = self._summarize_context(recent_context)
response = await self._context_aware_reasoning(
user_input,
context_summary,
confidence="medium"
)
return response
async def _process_hot(self, user_input: str) -> str:
"""Hot tier: Full context utilization with high confidence."""
context_summary = self._summarize_context(self.context_window)
preferences = self.preferences
response = await self._context_aware_reasoning(
user_input,
context_summary,
preferences=preferences,
confidence="high"
)
return response
async def _process_degraded(self, user_input: str) -> str:
"""Degraded tier: Context is stale, verify assumptions."""
verification = (
"It's been a while since our last interaction. "
"Let me verify that my understanding of your context "
"is still accurate before proceeding."
)
partial_response = await self._context_aware_reasoning(
user_input,
self._summarize_context(self.context_window[-5:]),
confidence="low"
)
return f"{verification}\n\n{partial_response}"
def _recalculate_tier(self) -> None:
"""Progressively upgrade tier based on interaction depth."""
if self.interaction_count >= self.HOT_THRESHOLD:
self.tier = AgentTier.HOT
elif self.interaction_count >= self.WARMING_THRESHOLD:
self.tier = AgentTier.WARMING
async def _rehydrate_context(self) -> None:
"""Attempt to restore context from persistent storage."""
# In production, this would query a session database
# to reload user preferences and recent history
pass
async def _generic_reasoning(self, user_input: str) -> str:
"""Fallback reasoning with no user-specific context."""
return "[Generic reasoning based on system knowledge]"
async def _context_aware_reasoning(self, user_input: str,
context: str,
preferences: Optional[Dict] = None,
confidence: str = "medium") -> str:
"""Reasoning enriched with accumulated context."""
return f"[{confidence} confidence response using context: {context[:100]}...]"
def _summarize_context(self, window: list) -> str:
"""Create a condensed summary of recent interactions."""
if not window:
return "No prior context available."
interactions = [f"User asked: {item['input']}" for item in window]
return "; ".join(interactions[-5:])
6. User Profiling and Session Persistence
The most impactful long-term solution to cold start is persistent user profiles that survive across sessions. When a user returns, the agent hydrates its state from durable storage rather than starting from zero. This requires careful design of the serialization format and a strategy for handling profile staleness.
# session_persistence.py
import json
import hashlib
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from dataclasses import dataclass, asdict, field
@dataclass
class UserProfile:
"""Persistent user profile that survives sessions."""
user_id: str
created_at: str
last_updated: str
explicit_preferences: Dict[str, Any] = field(default_factory=dict)
inferred_preferences: Dict[str, float] = field(default_factory=dict)
common_queries: list = field(default_factory=list)
successful_tools: Dict[str, int] = field(default_factory=dict)
failed_tools: Dict[str, int] = field(default_factory=dict)
domain_keywords: list = field(default_factory=list)
interaction_count: int = 0
last_session_summary: Optional[str] = None
class SessionManager:
"""Manages user session persistence and profile hydration
to eliminate cold starts on returning users."""
def __init__(self, storage_backend):
self.storage = storage_backend # Could be Redis, DynamoDB, PostgreSQL
self.active_sessions: Dict[str, UserProfile] = {}
self.profile_cache_ttl = timedelta(hours=1)
async def get_or_create_profile(self, user_id: str) -> UserProfile:
"""Hydrate user profile from persistent storage, or create new.
This is the primary cold-start mitigation entry point."""
# Check in-memory cache first
if user_id in self.active_sessions:
profile = self.active_sessions[user_id]
if self._is_profile_fresh(profile):
return profile
# Attempt persistent storage lookup
stored = await self.storage.get(f"profile:{user_id}")
if stored:
profile_data = json.loads(stored)
profile = UserProfile(**profile_data)
print(f"Hydrated profile for {user_id}: "
f"{profile.interaction_count} previous interactions, "
f"{len(profile.explicit_preferences)} explicit preferences")
self.active_sessions[user_id] = profile
return profile
# True cold start: create a new profile
new_profile = UserProfile(
user_id=user_id,
created_at=datetime.now().isoformat(),
last_updated=datetime.now().isoformat()
)
await self._persist_profile(new_profile)
self.active_sessions[user_id] = new_profile
print(f"Created new profile for {user_id} — cold start")
return new_profile
async def update_profile(self, user_id: str,
interaction: Dict[str, Any]) -> None:
"""Incrementally update profile based on interaction outcomes."""
profile = await self.get_or_create_profile(user_id)
profile.interaction_count += 1
profile.last_updated = datetime.now().isoformat()
# Update inferred preferences based on interaction
if "user_feedback" in interaction:
feedback = interaction["user_feedback"]
if isinstance(feedback, dict):
for key, value in feedback.items():
current = profile.inferred_preferences.get(key, 0.0)
# Exponential moving average with alpha=0.3
profile.inferred_preferences[key] = current * 0.7 + value * 0.3
# Track tool success/failure rates
if "tool_used" in interaction:
tool = interaction["tool_used"]
if interaction.get("tool_success", False):
profile.successful_tools[tool] = profile.successful_tools.get(tool, 0) + 1
else:
profile.failed_tools[tool] = profile.failed_tools.get(tool, 0) + 1
# Extract domain keywords from queries
if "query" in interaction:
keywords = self._extract_keywords(interaction["query"])
profile.domain_keywords.extend(keywords)
# Deduplicate and keep top 100
profile.domain_keywords = list(set(profile.domain_keywords))[:100]
profile.common_queries.append(interaction["query"])
if len(profile.common_queries) > 50:
profile.common_queries = profile.common_queries[-50:]
# Persist updated profile
await self._persist_profile(profile)
async def _persist_profile(self, profile: UserProfile) -> None:
"""Serialize and store profile to durable storage."""
profile_data = asdict(profile)
await self.storage.set(
f"profile:{profile.user_id}",
json.dumps(profile_data),
ttl=int(timedelta(days=30).total_seconds())
)
def _is_profile_fresh(self, profile: UserProfile) -> bool:
"""Check if cached profile is recent enough to use directly."""
last_update = datetime.fromisoformat(profile.last_updated)
return datetime.now() - last_update < self.profile_cache_ttl
def _extract_keywords(self, text: str) -> list:
"""Simple keyword extraction (in production, use a proper NLP pipeline)."""
# Remove stopwords and extract meaningful tokens
stopwords = {"a", "an", "the", "is", "are", "was", "were", "in", "on", "to", "for"}
words = text.lower().split()
return [w for w in words if w not in stopwords and len(w) > 3]
async def get_tool_recommendation(self, user_id: str, task: str) -> str:
"""Recommend the best tool based on user's historical success rates.
This converts cold start ignorance into data-driven selection."""
profile = await self.get_or_create_profile(user_id)
# Combine success and failure counts to compute reliability scores
tool_scores = {}
all_tools = set(list(profile.successful_tools.keys()) +
list(profile.failed_tools.keys()))
for tool in all_tools:
successes = profile.successful_tools.get(tool, 0)
failures = profile.failed_tools.get(tool, 0)
total = successes + failures
if total > 0:
# Wilson score lower bound for reliability
tool_scores[tool] = (successes + 1) / (total + 2)
else:
tool_scores[tool] = 0.5 # Neutral prior
if tool_scores:
best_tool = max(tool_scores, key=tool_scores.get)
return best_tool
return "fallback_tool" # No history available