AI Agents
Cortex provides the real-time knowledge layer that transforms your AI agents from static responders into dynamic, web-aware autonomous systems. Give your agents the ability to access, verify, and act on current information from across the web.
Why AI Agents Need Cortex
The Agent Knowledge Problem
- Stale Training Data: Agents trained on outdated information
- Hallucination Risk: Agents making up facts when knowledge is missing
- No Source Verification: Unable to cite or validate information
- Limited Context: Agents can't access real-time events or data
Cortex Solution
- Live Web Access: Real-time information retrieval
- Source Citations: Verifiable, traceable information
- Truth Validation: Cross-reference multiple sources
- Secure Processing: Filtered, safe content for agent consumption
Agent Architecture with Cortex
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ User Query │───▶│ AI Agent │───▶│ Cortex API │
│ │ │ (LangChain) │ │ │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Agent Action │◀───│ Reasoning │◀───│ Web Knowledge │
│ │ │ + Decision │ │ + Sources │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Implementation Examples
LangChain Integration
from langchain.agents import create_openai_functions_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
import cortex
# Initialize Cortex
cortex_client = cortex.Client(api_key="your_api_key")
@tool
def search_web(query: str) -> str:
"""Search the web for current information"""
result = cortex_client.search(
query=query,
max_results=5,
include_sources=True
)
# Format for agent consumption
response = f"Summary: {result.summary}\n\nSources:\n"
for source in result.sources:
response += f"- {source.title}: {source.url}\n"
return response
@tool
def verify_information(claim: str) -> str:
"""Verify a claim against multiple web sources"""
result = cortex_client.validate(
claim=claim,
sources=3,
confidence_threshold=0.8
)
return f"Verification: {result.verdict} (confidence: {result.confidence})"
# Create agent with Cortex tools
tools = [search_web, verify_information]
agent = create_openai_functions_agent(
llm=ChatOpenAI(model="gpt-4"),
tools=tools,
system_message="You are a helpful assistant with access to real-time web information."
)
AutoGPT-style Agent
import cortex
from typing import List, Dict
class WebAwareAgent:
def __init__(self, api_key: str):
self.cortex = cortex.Client(api_key=api_key)
self.memory = []
def think(self, goal: str) -> str:
"""Agent reasoning with web knowledge"""
# Get current information
context = self.cortex.search(
query=f"current information about {goal}",
max_results=10
)
# Reasoning prompt
prompt = f"""
Goal: {goal}
Current Web Context:
{context.summary}
Sources: {[s.url for s in context.sources]}
Based on this current information, what should I do next?
"""
return self.generate_response(prompt)
def act(self, action: str) -> Dict:
"""Execute action with web verification"""
# Verify action is still relevant
verification = self.cortex.validate(
claim=f"The action '{action}' is appropriate for current situation",
sources=3
)
if verification.confidence > 0.7:
return self.execute_action(action)
else:
return {"status": "skipped", "reason": "Information outdated"}
def research(self, topic: str) -> Dict:
"""Deep research on a topic"""
# Multi-angle research
angles = [
f"latest developments in {topic}",
f"expert opinions on {topic}",
f"challenges and problems with {topic}",
f"future trends in {topic}"
]
research_results = {}
for angle in angles:
result = self.cortex.search(query=angle, max_results=5)
research_results[angle] = {
"summary": result.summary,
"sources": result.sources,
"confidence": result.confidence
}
return research_results
# Usage
agent = WebAwareAgent(api_key="your_cortex_key")
research = agent.research("artificial general intelligence")
next_action = agent.think("become an AI safety researcher")
CrewAI Integration
from crewai import Agent, Task, Crew
from crewai_tools import BaseTool
import cortex
class CortexSearchTool(BaseTool):
name: str = "Web Search"
description: str = "Search the web for current information"
def __init__(self):
super().__init__()
self.cortex = cortex.Client(api_key="your_api_key")
def _run(self, query: str) -> str:
result = self.cortex.search(query=query, max_results=5)
return f"{result.summary}\n\nSources: {', '.join([s.url for s in result.sources])}"
class CortexValidationTool(BaseTool):
name: str = "Fact Verification"
description: str = "Verify information against multiple sources"
def __init__(self):
super().__init__()
self.cortex = cortex.Client(api_key="your_api_key")
def _run(self, claim: str) -> str:
result = self.cortex.validate(claim=claim)
return f"Verification: {result.verdict} (confidence: {result.confidence})"
# Create specialized agents
researcher = Agent(
role='Research Analyst',
goal='Gather and analyze current information',
backstory='Expert at finding and synthesizing current information',
tools=[CortexSearchTool(), CortexValidationTool()],
verbose=True
)
fact_checker = Agent(
role='Fact Checker',
goal='Verify information accuracy',
backstory='Specialist in validating claims against multiple sources',
tools=[CortexValidationTool()],
verbose=True
)
# Create tasks
research_task = Task(
description='Research the latest developments in quantum computing',
agent=researcher,
expected_output='Comprehensive report with verified facts and sources'
)
verification_task = Task(
description='Verify all claims in the research report',
agent=fact_checker,
expected_output='Fact-checked report with confidence scores'
)
# Create crew
crew = Crew(
agents=[researcher, fact_checker],
tasks=[research_task, verification_task],
verbose=2
)
result = crew.kickoff()
Agent Patterns
Research Agent
class ResearchAgent:
def __init__(self, cortex_client):
self.cortex = cortex_client
self.findings = []
def investigate(self, topic: str) -> Dict:
"""Comprehensive topic investigation"""
# Phase 1: Broad search
overview = self.cortex.search(
query=f"overview of {topic}",
max_results=10
)
# Phase 2: Specific angles
angles = self.extract_angles(overview.summary)
detailed_findings = []
for angle in angles:
result = self.cortex.search(
query=f"{topic} {angle}",
max_results=5
)
detailed_findings.append(result)
# Phase 3: Verification
key_facts = self.extract_key_facts(detailed_findings)
verified_facts = []
for fact in key_facts:
verification = self.cortex.validate(claim=fact)
if verification.confidence > 0.8:
verified_facts.append({
"fact": fact,
"confidence": verification.confidence,
"sources": verification.sources
})
return {
"topic": topic,
"overview": overview.summary,
"verified_facts": verified_facts,
"total_sources": len(set([s.url for finding in detailed_findings for s in finding.sources]))
}
Decision Support Agent
class DecisionAgent:
def __init__(self, cortex_client):
self.cortex = cortex_client
def analyze_decision(self, decision: str, context: str) -> Dict:
"""Analyze a decision with current information"""
# Gather relevant information
pro_info = self.cortex.search(
query=f"benefits advantages of {decision} in {context}",
max_results=5
)
con_info = self.cortex.search(
query=f"risks disadvantages problems with {decision} in {context}",
max_results=5
)
# Get expert opinions
expert_opinions = self.cortex.search(
query=f"expert opinion analysis {decision} {context}",
max_results=5
)
# Validate key claims
key_claims = self.extract_claims([pro_info, con_info, expert_opinions])
validated_claims = []
for claim in key_claims:
validation = self.cortex.validate(claim=claim)
validated_claims.append({
"claim": claim,
"validated": validation.confidence > 0.7,
"confidence": validation.confidence
})
return {
"decision": decision,
"context": context,
"pros": pro_info.summary,
"cons": con_info.summary,
"expert_insights": expert_opinions.summary,
"validated_claims": validated_claims,
"recommendation": self.generate_recommendation(validated_claims)
}
Monitoring Agent
class MonitoringAgent:
def __init__(self, cortex_client):
self.cortex = cortex_client
self.watched_topics = []
self.alerts = []
def add_watch(self, topic: str, alert_threshold: float = 0.3):
"""Add a topic to monitor"""
# Get baseline information
baseline = self.cortex.search(
query=topic,
max_results=5
)
self.watched_topics.append({
"topic": topic,
"baseline": baseline,
"alert_threshold": alert_threshold,
"last_check": datetime.now()
})
def check_updates(self):
"""Check for updates on watched topics"""
for watch in self.watched_topics:
current = self.cortex.search(
query=watch["topic"],
max_results=5
)
# Use change monitoring
changes = self.cortex.monitor_changes(
baseline_sources=[s.url for s in watch["baseline"].sources],
current_sources=[s.url for s in current.sources]
)
if changes.significance_score > watch["alert_threshold"]:
self.alerts.append({
"topic": watch["topic"],
"change_type": changes.change_type,
"significance": changes.significance_score,
"summary": changes.summary,
"timestamp": datetime.now()
})
def get_alerts(self) -> List[Dict]:
"""Get pending alerts"""
return self.alerts
Advanced Features
Multi-Source Validation
def validate_with_consensus(claim: str, min_sources: int = 3) -> Dict:
"""Validate a claim using multiple sources"""
validation = cortex_client.validate(
claim=claim,
min_sources=min_sources,
consensus_threshold=0.8
)
return {
"claim": claim,
"validated": validation.consensus_reached,
"confidence": validation.confidence,
"supporting_sources": validation.supporting_sources,
"contradicting_sources": validation.contradicting_sources,
"neutral_sources": validation.neutral_sources
}
Temporal Reasoning
def get_temporal_context(event: str, timeframe: str) -> Dict:
"""Get temporal context for an event"""
result = cortex_client.search(
query=f"{event} {timeframe}",
time_filter={
"start": timeframe,
"end": "now"
},
sort_by="recency"
)
return {
"event": event,
"timeframe": timeframe,
"timeline": result.timeline,
"recent_developments": result.summary,
"trend_analysis": result.trend_analysis
}
Source Quality Assessment
def assess_source_quality(sources: List[str]) -> Dict:
"""Assess the quality and reliability of sources"""
assessment = cortex_client.assess_sources(
urls=sources,
criteria=["authority", "accuracy", "currency", "objectivity"]
)
return {
"high_quality": [s for s in assessment if s.score > 0.8],
"medium_quality": [s for s in assessment if 0.5 < s.score <= 0.8],
"low_quality": [s for s in assessment if s.score <= 0.5],
"average_score": sum(s.score for s in assessment) / len(assessment)
}
Best Practices
Error Handling
def robust_agent_search(query: str, max_retries: int = 3) -> Dict:
"""Robust search with error handling"""
for attempt in range(max_retries):
try:
result = cortex_client.search(query=query)
return {
"success": True,
"result": result,
"attempt": attempt + 1
}
except cortex.RateLimitError:
time.sleep(2 ** attempt) # Exponential backoff
except cortex.SecurityError as e:
return {
"success": False,
"error": "Security filter triggered",
"suggestion": "Rephrase query to avoid potential security issues"
}
except Exception as e:
if attempt == max_retries - 1:
return {
"success": False,
"error": str(e),
"fallback": "Use cached knowledge or inform user of limitation"
}
return {"success": False, "error": "Max retries exceeded"}
Performance Optimization
class OptimizedAgent:
def __init__(self, cortex_client):
self.cortex = cortex_client
self.cache = {}
self.cache_duration = 300 # 5 minutes
def cached_search(self, query: str) -> Dict:
"""Search with caching"""
cache_key = hashlib.md5(query.encode()).hexdigest()
if cache_key in self.cache:
cached_result = self.cache[cache_key]
if time.time() - cached_result["timestamp"] < self.cache_duration:
return cached_result["result"]
result = self.cortex.search(query=query)
self.cache[cache_key] = {
"result": result,
"timestamp": time.time()
}
return result
def batch_search(self, queries: List[str]) -> List[Dict]:
"""Batch multiple searches for efficiency"""
return self.cortex.batch_search(queries=queries)
Agent Memory Integration
class MemoryAwareAgent:
def __init__(self, cortex_client):
self.cortex = cortex_client
self.long_term_memory = []
self.working_memory = []
def remember_fact(self, fact: str, source: str, confidence: float):
"""Store verified facts in long-term memory"""
self.long_term_memory.append({
"fact": fact,
"source": source,
"confidence": confidence,
"timestamp": datetime.now(),
"verified": True
})
def recall_relevant_facts(self, query: str) -> List[Dict]:
"""Recall relevant facts from memory"""
# Simple similarity matching (use embeddings in production)
relevant_facts = []
for memory in self.long_term_memory:
if any(word in memory["fact"].lower() for word in query.lower().split()):
relevant_facts.append(memory)
return sorted(relevant_facts, key=lambda x: x["confidence"], reverse=True)
def hybrid_search(self, query: str) -> Dict:
"""Combine memory and web search"""
# Check memory first
memory_results = self.recall_relevant_facts(query)
# Get fresh web information
web_results = self.cortex.search(query=query)
# Combine and validate
combined_info = {
"query": query,
"memory_facts": memory_results,
"web_info": web_results,
"synthesis": self.synthesize_information(memory_results, web_results)
}
return combined_info
Production Deployment
Scaling Considerations
# Use connection pooling for high-volume agents
cortex_client = cortex.Client(
api_key="your_api_key",
connection_pool_size=20,
timeout=30,
retry_config={
"max_retries": 3,
"backoff_factor": 2
}
)
# Implement circuit breaker pattern
class CircuitBreakerAgent:
def __init__(self, cortex_client):
self.cortex = cortex_client
self.failure_count = 0
self.failure_threshold = 5
self.recovery_timeout = 60
self.last_failure_time = None
self.state = "closed" # closed, open, half-open
def search_with_circuit_breaker(self, query: str):
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "half-open"
else:
return {"error": "Circuit breaker is open"}
try:
result = self.cortex.search(query=query)
if self.state == "half-open":
self.state = "closed"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"
raise e
Next: Chatbots → -