External Knowledge Integration - Summary¶

# Target performance: <100ms end-to-end

def get_knowledge(context: str) -> Optional[Dict]:
    """Optimized knowledge retrieval."""

    # Stage 1: Project memory (5-10ms)
    project_mem = memory_manager.retrieve(search=context, limit=1)
    if project_mem:
        return project_mem[0]

    # Stage 2: Cache lookup (10-20ms)
    cached = cache.get(source, identifier)
    if cached:
        return cached

    # Stage 3: External fetch (50-500ms)
    # Only if really needed
    if should_fetch_external(context):
        return fetch_and_cache(source, identifier)

    return None

def should_refresh(doc: ExternalDoc) -> bool:
    """Smart refresh logic."""

    # Refresh if:
    # 1. Older than TTL
    # 2. High value (top 20% by access count)
    # 3. Recently used (last 7 days)

    is_stale = (now - doc.last_updated) > TTL[doc.category]
    is_valuable = doc.access_count > percentile_80
    recently_used = (now - doc.last_accessed).days < 7

    return is_stale and (is_valuable or recently_used)

def detect_deprecation(api_ref: APIReference) -> Optional[Dict]:
    """Check if API has been deprecated."""

    # Methods:
    # 1. Parse official docs for deprecation notices
    # 2. Check library CHANGELOG
    # 3. Monitor GitHub issues

    doc = fetch_official_docs(api_ref.namespace)
    if "deprecated" in doc.lower():
        return extract_deprecation_info(doc)

    return None

def calculate_relevance_score(doc: ExternalDoc, context: str) -> float:
    """
    Calculate relevance based on multiple factors.

    Weights:
    - Source credibility: 40%
    - Content freshness: 20%
    - Usage frequency: 20%
    - Text similarity: 20%
    """

    credibility = SOURCE_TRUST_SCORES[doc.source]
    freshness = 1.0 - (days_old / 730.0)  # 2-year decay
    usage = min(1.0, doc.access_count / 100.0)
    similarity = text_similarity(doc.summary, context)

    return (
        credibility * 0.40 +
        freshness * 0.20 +
        usage * 0.20 +
        similarity * 0.20
    )

# BEFORE: Agent context from project memory only
context = memory_manager.retrieve(agent_id, search=task)

# AFTER: Agent context from project memory + external knowledge
def build_agent_context(agent_id: str, task: str) -> str:
    """Build context from multiple sources."""

    context_parts = []

    # 1. Project memory (ALWAYS FIRST)
    project_memories = memory_manager.retrieve(
        agent_id=agent_id,
        search=task,
        min_importance=5
    )
    if project_memories:
        context_parts.append("## Project-Specific Knowledge")
        for mem in project_memories:
            context_parts.append(f"- {mem.title}: {mem.content}")

    # 2. External knowledge (IF NEEDED)
    if external_retriever.should_query_external(task):
        external_docs = external_retriever.get_relevant_docs(task, limit=2)
        if external_docs:
            context_parts.append("\n## External Reference (Advisory)")
            for doc in external_docs:
                context_parts.append(f"- [{doc.source}] {doc.title}: {doc.summary}")

    return "\n".join(context_parts)

# Phase 1: File cache only
cache = ExternalKnowledgeCache()
doc = cache.get("python_docs", "asyncio.run")

# Phase 2: Add memory integration (backwards compatible)
retriever = ExternalKnowledgeRetriever(memory_manager)
doc = retriever.get_function_doc("python", "asyncio", "run")
# Still uses file cache, but stores in memory too

# Only migrate if measurements show need

if cache_hit_rate < 0.7 or avg_query_time > 100:
    # Add Neo4j for metadata
    neo4j = ExternalKnowledgeNeo4j(uri, user, password)

    # Migrate existing cache metadata to Neo4j
    migrate_cache_to_neo4j(cache, neo4j)

    # Keep file cache for full content
    # Neo4j for fast metadata queries

def daily_maintenance():
    """Automated daily tasks."""

    # 1. Refresh high-value cached docs
    refresh_docs_if_needed(access_count_percentile=0.8)

    # 2. Clean up old cache entries
    cleanup_cache(older_than_days=90, unused=True)

    # 3. Update relevance scores
    recalculate_relevance_scores()

def weekly_analysis():
    """Generate usage reports."""

    return {
        "cache_hit_rate": calculate_hit_rate(),
        "most_used_docs": get_top_documents(20),
        "sources_by_usage": analyze_source_effectiveness(),
        "knowledge_gaps": identify_gaps(),
        "avg_query_time_ms": get_avg_query_time()
    }