Agentic Retrieval

# Complex context gathering
"What context would help me understand the evolution of our pricing strategy?"

# Multi-faceted questions
"What are all the factors we discussed that affect the product launch timeline?"

# Relationship queries
"What connections exist between our customer feedback and the recent product changes?"

# Preparatory context
"Help me prepare for a conversation with the engineering team about technical debt"

# Simple lookups
"What did Alice say about the budget?"

# Direct searches
"Find discussions about Kubernetes"

# Recent context
"What did we discuss in yesterday's meeting?"

import requests
from datetime import datetime

BASE_URL = "https://api.evermind.ai"
headers = {"Content-Type": "application/json"}

def agentic_search(user_id: str, query: str, top_k: int = 10) -> dict:
    """Perform agentic retrieval."""
    search_params = {
        "user_id": user_id,
        "query": query,
        "retrieve_method": "agentic",  # Enable agentic retrieval
        "top_k": top_k,
        "memory_types": ["episodic_memory", "profile"]
    }

    response = requests.get(
        f"{BASE_URL}/api/v0/memories/search",
        json=search_params,
        headers=headers,
        timeout=60  # Longer timeout for agentic
    )

    return response.json()

# Example usage
result = agentic_search(
    user_id="user_alice",
    query="What context would help me prepare for discussing the product roadmap with stakeholders?"
)

memories = result.get("result", {}).get("memories", [])
print(f"Found {len(memories)} relevant memories")
for mem in memories:
    print(f"- {mem.get('memory_content', '')[:100]}...")

def prepare_meeting_context(user_id: str, meeting_topic: str, attendees: list) -> dict:
    """Gather comprehensive context for a meeting."""

    # Build query incorporating topic and attendees
    attendee_names = ", ".join(attendees)
    query = f"""What context would help me prepare for a meeting about {meeting_topic}?
    Attendees include: {attendee_names}.
    I need:
    - Previous discussions on this topic
    - Relevant decisions and outcomes
    - Any concerns or blockers mentioned
    - Attendee preferences and working styles"""

    search_params = {
        "user_id": user_id,
        "query": query,
        "retrieve_method": "agentic",
        "top_k": 15,
        "memory_types": ["episodic_memory", "profile"]
    }

    response = requests.get(
        f"{BASE_URL}/api/v0/memories/search",
        json=search_params,
        headers=headers,
        timeout=60
    )

    return response.json()

# Usage
context = prepare_meeting_context(
    user_id="user_alice",
    meeting_topic="Q3 product roadmap",
    attendees=["Bob (Engineering)", "Carol (Product)", "Dave (Sales)"]
)

def trace_decision_history(user_id: str, decision_topic: str) -> dict:
    """Trace the evolution of decisions on a topic."""

    query = f"""Trace the history of decisions and discussions about {decision_topic}.
    I want to understand:
    - What options were considered
    - What factors influenced the decisions
    - Who was involved in the discussions
    - What the outcomes were
    - Any changes or reversals over time"""

    search_params = {
        "user_id": user_id,
        "query": query,
        "retrieve_method": "agentic",
        "top_k": 20,
        "memory_types": ["episodic_memory"]
    }

    response = requests.get(
        f"{BASE_URL}/api/v0/memories/search",
        json=search_params,
        headers=headers,
        timeout=60
    )

    return response.json()

# Usage
history = trace_decision_history(
    user_id="user_alice",
    decision_topic="choosing our cloud provider"
)

def analyze_topic_relationships(user_id: str, topics: list) -> dict:
    """Find connections between multiple topics."""

    topics_str = ", ".join(topics)
    query = f"""Find connections and relationships between these topics: {topics_str}.
    Look for:
    - How these topics have been discussed together
    - Dependencies or conflicts between them
    - People involved in multiple topics
    - Timeline overlaps"""

    search_params = {
        "user_id": user_id,
        "query": query,
        "retrieve_method": "agentic",
        "top_k": 15,
        "memory_types": ["episodic_memory"]
    }

    response = requests.get(
        f"{BASE_URL}/api/v0/memories/search",
        json=search_params,
        headers=headers,
        timeout=60
    )

    return response.json()

# Usage
relationships = analyze_topic_relationships(
    user_id="user_alice",
    topics=["customer feedback", "product features", "technical debt"]
)

# 1. Use appropriate top_k - don't over-request
search_params = {
    "top_k": 10,  # Enough context, not excessive
    ...
}

# 2. Filter memory types when possible
search_params = {
    "memory_types": ["episodic_memory"],  # Only what's needed
    ...
}

# 3. Use group_ids for scoped searches
search_params = {
    "group_ids": ["specific_conversation"],  # Narrow the search space
    ...
}

import asyncio

async def tiered_retrieval(user_id: str, query: str, complexity: str = "auto") -> dict:
    """Use appropriate retrieval based on query complexity."""

    if complexity == "auto":
        complexity = estimate_complexity(query)

    if complexity == "simple":
        # Fast path for simple queries
        method = "hybrid"
        timeout = 10
    else:
        # Complex queries get agentic
        method = "agentic"
        timeout = 60

    search_params = {
        "user_id": user_id,
        "query": query,
        "retrieve_method": method,
        "top_k": 10,
        "memory_types": ["episodic_memory", "profile"]
    }

    try:
        response = requests.get(
            f"{BASE_URL}/api/v0/memories/search",
            json=search_params,
            headers=headers,
            timeout=timeout
        )
        return response.json()

    except requests.Timeout:
        if method == "agentic":
            # Fallback to hybrid on timeout
            search_params["retrieve_method"] = "hybrid"
            response = requests.get(
                f"{BASE_URL}/api/v0/memories/search",
                json=search_params,
                headers=headers,
                timeout=10
            )
            return response.json()
        raise


def estimate_complexity(query: str) -> str:
    """Estimate if a query needs agentic retrieval."""
    complex_indicators = [
        "prepare", "context", "help me understand", "trace",
        "relationship", "connection", "evolution", "history of",
        "factors", "all the", "comprehensive"
    ]

    query_lower = query.lower()

    # Check for complexity indicators
    if any(indicator in query_lower for indicator in complex_indicators):
        return "complex"

    # Check for multi-part queries
    if query.count("?") > 1 or " and " in query_lower:
        return "complex"

    return "simple"

import aiohttp
import asyncio

class AgenticEverOSClient:
    """Client optimized for agentic retrieval."""

    def __init__(self, base_url: str = "https://api.evermind.ai"):
        self.base_url = base_url
        self.headers = {"Content-Type": "application/json"}

    async def agentic_search(
        self,
        user_id: str,
        query: str,
        top_k: int = 10,
        timeout: int = 60
    ) -> dict:
        """Perform agentic search with appropriate timeout."""

        search_params = {
            "user_id": user_id,
            "query": query,
            "retrieve_method": "agentic",
            "top_k": top_k,
            "memory_types": ["episodic_memory", "profile"]
        }

        client_timeout = aiohttp.ClientTimeout(total=timeout)

        async with aiohttp.ClientSession(timeout=client_timeout) as session:
            async with session.get(
                f"{self.base_url}/api/v0/memories/search",
                json=search_params,
                headers=self.headers
            ) as response:
                response.raise_for_status()
                return await response.json()

    async def search_with_fallback(
        self,
        user_id: str,
        query: str,
        prefer_agentic: bool = True
    ) -> dict:
        """Search with automatic fallback."""

        if prefer_agentic:
            try:
                return await self.agentic_search(user_id, query, timeout=60)
            except asyncio.TimeoutError:
                pass  # Fall through to hybrid

        # Hybrid fallback
        search_params = {
            "user_id": user_id,
            "query": query,
            "retrieve_method": "hybrid",
            "top_k": 10,
            "memory_types": ["episodic_memory", "profile"]
        }

        client_timeout = aiohttp.ClientTimeout(total=10)
        async with aiohttp.ClientSession(timeout=client_timeout) as session:
            async with session.get(
                f"{self.base_url}/api/v0/memories/search",
                json=search_params,
                headers=self.headers
            ) as response:
                response.raise_for_status()
                return await response.json()


# Usage
async def main():
    client = AgenticEverOSClient()

    result = await client.search_with_fallback(
        user_id="user_alice",
        query="What context do I need to understand our pricing strategy evolution?",
        prefer_agentic=True
    )

    print(f"Found {len(result.get('result', {}).get('memories', []))} memories")

asyncio.run(main())