"""
Pattern 126: Knowledge Fusion

Description:
    Combines knowledge from multiple heterogeneous sources, resolving conflicts
    and creating a unified, consistent knowledge representation.

Components:
    - Multi-source integration
    - Conflict detection and resolution
    - Consistency checking
    - Knowledge merging strategies

Use Cases:
    - Multi-source data integration
    - Fact verification across sources
    - Building comprehensive knowledge bases

LangChain Implementation:
    Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.
"""

import os
from typing import List, Dict, Any, Optional, Set
from dotenv import load_dotenv
from datetime import datetime
from collections import defaultdict

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.pydantic_v1 import BaseModel, Field

load_dotenv()


class KnowledgeSource(BaseModel):
    """Represents a knowledge source"""
    source_id: str = Field(description="Unique source identifier")
    name: str = Field(description="Source name")
    reliability: float = Field(description="Source reliability score 0-1")
    data: Dict[str, Any] = Field(description="Knowledge data")


class FusedKnowledge(BaseModel):
    """Fused knowledge item"""
    key: str = Field(description="Knowledge key/identifier")
    value: Any = Field(description="Fused value")
    confidence: float = Field(description="Confidence in fused value")
    sources: List[str] = Field(description="Contributing sources")
    conflicts: List[Dict[str, Any]] = Field(description="Detected conflicts", default_factory=list)


class KnowledgeFusionAgent:
    """Agent for fusing knowledge from multiple sources"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.0):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.sources: List[KnowledgeSource] = []
        self.fused_knowledge: Dict[str, FusedKnowledge] = {}
        
        # Conflict resolution prompt
        self.conflict_resolution_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert at resolving conflicting information from multiple sources.
Given conflicting facts from different sources with their reliability scores, determine:
1. Which fact is most likely correct
2. If a synthesis/combination is possible
3. Confidence level in the resolution

Return JSON with format:
{{
  "resolved_value": "the resolved/synthesized value",
  "confidence": 0.85,
  "reasoning": "explanation of resolution",
  "method": "voting|weighted|synthesis|newest"
}}"""),
            ("user", """Conflicting information about: {key}

Sources and values:
{conflicts}

Resolve the conflict.""")
        ])
        
        self.parser = JsonOutputParser()
    
    def add_source(self, source: KnowledgeSource):
        """Add a knowledge source"""
        self.sources.append(source)
        print(f"✓ Added source: {source.name} (reliability: {source.reliability})")
    
    def detect_conflicts(self, key: str, values: List[Dict[str, Any]]) -> bool:
        """Check if there are conflicts for a key"""
        if len(values)  1
    
    def simple_fusion(self, key: str, values: List[Dict[str, Any]]) -> FusedKnowledge:
        """Simple fusion without conflicts (voting or consensus)"""
        if len(values) == 1:
            # Single source - use as is
            return FusedKnowledge(
                key=key,
                value=values[0]['value'],
                confidence=values[0]['reliability'],
                sources=[values[0]['source_id']]
            )
        
        # Multiple sources with same value - average confidence
        avg_confidence = sum(v['reliability'] for v in values) / len(values)
        sources = [v['source_id'] for v in values]
        
        return FusedKnowledge(
            key=key,
            value=values[0]['value'],
            confidence=avg_confidence,
            sources=sources
        )
    
    def resolve_conflict(self, key: str, values: List[Dict[str, Any]]) -> FusedKnowledge:
        """Resolve conflicting values using LLM"""
        print(f"\n⚠️  Conflict detected for '{key}'")
        
        # Prepare conflict description
        conflicts_text = ""
        conflict_list = []
        for v in values:
            conflicts_text += f"- Source '{v['source_name']}' (reliability: {v['reliability']}): {v['value']}\n"
            conflict_list.append({
                "source": v['source_name'],
                "value": v['value'],
                "reliability": v['reliability']
            })
        
        print(f"   Conflicting values:\n{conflicts_text}")
        
        try:
            # Use LLM to resolve
            chain = self.conflict_resolution_prompt | self.llm | self.parser
            result = chain.invoke({"key": key, "conflicts": conflicts_text})
            
            print(f"   Resolution: {result['resolved_value']}")
            print(f"   Method: {result['method']}, Confidence: {result['confidence']}")
            print(f"   Reasoning: {result['reasoning']}")
            
            return FusedKnowledge(
                key=key,
                value=result['resolved_value'],
                confidence=result['confidence'],
                sources=[v['source_id'] for v in values],
                conflicts=conflict_list
            )
        except Exception as e:
            print(f"   Error resolving conflict: {e}")
            # Fallback: use most reliable source
            best_source = max(values, key=lambda x: x['reliability'])
            return FusedKnowledge(
                key=key,
                value=best_source['value'],
                confidence=best_source['reliability'] * 0.8,  # Reduce confidence
                sources=[v['source_id'] for v in values],
                conflicts=conflict_list
            )
    
    def fuse_knowledge(self) -> Dict[str, FusedKnowledge]:
        """Fuse knowledge from all sources"""
        print("\n" + "=" * 80)
        print("🔄 Starting Knowledge Fusion Process")
        print("=" * 80)
        
        # Group knowledge by key
        knowledge_by_key = defaultdict(list)
        
        for source in self.sources:
            for key, value in source.data.items():
                knowledge_by_key[key].append({
                    'source_id': source.source_id,
                    'source_name': source.name,
                    'reliability': source.reliability,
                    'value': value
                })
        
        print(f"\n📊 Found {len(knowledge_by_key)} unique knowledge keys")
        
        # Fuse each key
        conflicts_detected = 0
        for key, values in knowledge_by_key.items():
            if self.detect_conflicts(key, values):
                conflicts_detected += 1
                self.fused_knowledge[key] = self.resolve_conflict(key, values)
            else:
                self.fused_knowledge[key] = self.simple_fusion(key, values)
        
        print(f"\n✓ Fusion complete!")
        print(f"  - Total keys: {len(self.fused_knowledge)}")
        print(f"  - Conflicts resolved: {conflicts_detected}")
        
        return self.fused_knowledge
    
    def query(self, key: str) -> Optional[FusedKnowledge]:
        """Query fused knowledge"""
        return self.fused_knowledge.get(key)
    
    def get_high_confidence_facts(self, threshold: float = 0.8) -> List[FusedKnowledge]:
        """Get facts with confidence above threshold"""
        return [
            fact for fact in self.fused_knowledge.values()
            if fact.confidence >= threshold
        ]
    
    def get_conflicted_facts(self) -> List[FusedKnowledge]:
        """Get facts that had conflicts"""
        return [
            fact for fact in self.fused_knowledge.values()
            if fact.conflicts
        ]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get fusion statistics"""
        total_facts = len(self.fused_knowledge)
        conflicted = len(self.get_conflicted_facts())
        high_confidence = len(self.get_high_confidence_facts())
        
        avg_confidence = sum(f.confidence for f in self.fused_knowledge.values()) / total_facts if total_facts > 0 else 0
        
        return {
            "total_sources": len(self.sources),
            "total_facts": total_facts,
            "conflicted_facts": conflicted,
            "high_confidence_facts": high_confidence,
            "average_confidence": avg_confidence
        }


def demonstrate_knowledge_fusion():
    """Demonstrate knowledge fusion from multiple sources"""
    print("=" * 80)
    print("Pattern 126: Knowledge Fusion")
    print("=" * 80)
    
    agent = KnowledgeFusionAgent()
    
    # Example 1: Add multiple knowledge sources
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Adding Knowledge Sources")
    print("=" * 80)
    
    # Source 1: Wikipedia (high reliability)
    wikipedia = KnowledgeSource(
        source_id="wikipedia",
        name="Wikipedia",
        reliability=0.85,
        data={
            "eiffel_tower_height": "330 meters",
            "eiffel_tower_built": "1889",
            "eiffel_tower_location": "Paris, France",
            "mount_everest_height": "8,849 meters",
            "python_creator": "Guido van Rossum"
        }
    )
    
    # Source 2: Tourist website (medium reliability)
    tourist_site = KnowledgeSource(
        source_id="tourist_site",
        name="Tourism Website",
        reliability=0.65,
        data={
            "eiffel_tower_height": "324 meters",  # Conflict!
            "eiffel_tower_built": "1889",
            "eiffel_tower_location": "Paris, France",
            "eiffel_tower_visitors": "7 million per year"
        }
    )
    
    # Source 3: News article (medium reliability)
    news = KnowledgeSource(
        source_id="news",
        name="News Article",
        reliability=0.70,
        data={
            "eiffel_tower_height": "330 meters including antenna",  # Conflict!
            "mount_everest_height": "8,848.86 meters",  # Conflict!
            "python_creator": "Guido van Rossum",
            "python_first_release": "1991"
        }
    )
    
    agent.add_source(wikipedia)
    agent.add_source(tourist_site)
    agent.add_source(news)
    
    # Example 2: Fuse knowledge
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Fusing Knowledge")
    print("=" * 80)
    
    fused = agent.fuse_knowledge()
    
    # Example 3: Query fused knowledge
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Querying Fused Knowledge")
    print("=" * 80)
    
    queries = [
        "eiffel_tower_height",
        "eiffel_tower_location",
        "python_creator",
        "mount_everest_height"
    ]
    
    for query in queries:
        result = agent.query(query)
        if result:
            print(f"\n🔍 Query: {query}")
            print(f"   Value: {result.value}")
            print(f"   Confidence: {result.confidence:.2f}")
            print(f"   Sources: {', '.join(result.sources)}")
            if result.conflicts:
                print(f"   ⚠️  Had conflicts from {len(result.conflicts)} sources")
    
    # Example 4: High confidence facts
    print("\n" + "=" * 80)
    print("EXAMPLE 4: High Confidence Facts")
    print("=" * 80)
    
    high_conf = agent.get_high_confidence_facts(threshold=0.75)
    print(f"\n✓ Found {len(high_conf)} high-confidence facts (>0.75):\n")
    for fact in high_conf:
        print(f"  - {fact.key}: {fact.value} (confidence: {fact.confidence:.2f})")
    
    # Example 5: Conflicted facts
    print("\n" + "=" * 80)
    print("EXAMPLE 5: Facts with Conflicts")
    print("=" * 80)
    
    conflicted = agent.get_conflicted_facts()
    print(f"\n⚠️  Found {len(conflicted)} facts with conflicts:\n")
    for fact in conflicted:
        print(f"  - {fact.key}:")
        print(f"      Resolved to: {fact.value}")
        print(f"      Original values:")
        for conflict in fact.conflicts:
            print(f"        • {conflict['source']}: {conflict['value']}")
    
    # Example 6: Statistics
    print("\n" + "=" * 80)
    print("EXAMPLE 6: Fusion Statistics")
    print("=" * 80)
    
    stats = agent.get_statistics()
    print(f"\n📊 Statistics:")
    print(f"  Total Sources: {stats['total_sources']}")
    print(f"  Total Facts: {stats['total_facts']}")
    print(f"  Conflicted Facts: {stats['conflicted_facts']}")
    print(f"  High Confidence Facts: {stats['high_confidence_facts']}")
    print(f"  Average Confidence: {stats['average_confidence']:.2f}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Knowledge Fusion Pattern:
- Integrates knowledge from multiple heterogeneous sources
- Detects and resolves conflicting information
- Uses source reliability scores for weighting
- Creates unified, consistent knowledge representation
- Maintains provenance and conflict information

Conflict Resolution Strategies:
✓ Voting (majority wins)
✓ Weighted by source reliability
✓ LLM-based synthesis
✓ Temporal (newest information)
✓ Consensus building

Key Benefits:
✓ Comprehensive knowledge coverage
✓ Improved accuracy through verification
✓ Conflict awareness and resolution
✓ Source tracking and attribution
✓ Confidence scoring
    """)


if __name__ == "__main__":
    demonstrate_knowledge_fusion()

    

Agentic Design Pattern: Knowledge Fusion

Pattern 126: Knowledge Fusion – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Knowledge Fusion pattern implementation.

---

Pattern Description

Pattern 126: Knowledge Fusion

Description: Combines knowledge from multiple heterogeneous sources, resolving conflicts and creating a unified, consistent knowledge representation.

Components: – Multi-source integration – Conflict detection and resolution – Consistency checking – Knowledge merging strategies

Use Cases: – Multi-source data integration – Fact verification across sources – Building comprehensive knowledge bases

LangChain Implementation: Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.

---

File Statistics

• Total Lines: 396
• Classes: 3
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 126: Knowledge Fusion

Description:
    Combines knowledge from multiple heterogeneous sources, resolving conflicts
    and creating a unified, consistent knowledge representation.

Components:
    - Multi-source integration
    - Conflict detection and resolution
    - Consistency checking
    - Knowledge merging strategies

Use Cases:
    - Multi-source data integration
    - Fact verification across sources
    - Building comprehensive knowledge bases

LangChain Implementation:
    Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.
"""

Purpose: This module implements the Knowledge Fusion pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.defaultdict
import datetime.datetime
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.pydantic_v1.BaseModel
import langchain_core.pydantic_v1.Field
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – datetime.datetime: Date and time handling – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.pydantic_v1.BaseModel: Regular expression operations – langchain_core.pydantic_v1.Field: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Set: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: KnowledgeSource

Lines 37-42

Inherits from: BaseModel

Purpose:

Represents a knowledge source

---

Class: FusedKnowledge

Lines 45-51

Inherits from: BaseModel

Purpose:

Fused knowledge item

---

Class: KnowledgeFusionAgent

Lines 54-240

Purpose:

Agent for fusing knowledge from multiple sources

Methods:

__init__(self, model, temperature)

Lines 57-85

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

add_source(self, source)

Lines 87-90

Purpose:

Add a knowledge source

Parameters: – source: Source

detect_conflicts(self, key, values)

Lines 92-104

Purpose:

Check if there are conflicts for a key

Parameters: – key: Key – values: Values

Returns: bool

simple_fusion(self, key, values)

Lines 106-126

Purpose:

Simple fusion without conflicts (voting or consensus)

Parameters: – key: Key – values: Values

Returns: FusedKnowledge

resolve_conflict(self, key, values)

Lines 128-171

Purpose:

Resolve conflicting values using LLM

Parameters: – key: Key – values: Values

Returns: FusedKnowledge

fuse_knowledge(self)

Lines 173-206

Purpose:

Fuse knowledge from all sources

Parameters:

Returns: <ast.Subscript object at 0x000001A06D88A210>

query(self, key)

Lines 208-210

Purpose:

Query fused knowledge

Parameters: – key: Key

Returns: <ast.Subscript object at 0x000001A06D889D10>

get_high_confidence_facts(self, threshold)

Lines 212-217

Purpose:

Get facts with confidence above threshold

Parameters: – threshold: Threshold

Returns: <ast.Subscript object at 0x000001A06D64A710>

get_conflicted_facts(self)

Lines 219-224

Purpose:

Get facts that had conflicts

Parameters:

Returns: <ast.Subscript object at 0x000001A06D64BD10>

get_statistics(self)

Lines 226-240

Purpose:

Get fusion statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D748190>

---

Functions

demonstrate_knowledge_fusion()

Lines 243-391

Purpose:

Demonstrate knowledge fusion from multiple sources

---

Key Concepts Summary

Pattern Overview

The Knowledge Fusion pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 125
• Pattern 127

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Knowledge Fusion pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 126_knowledge_fusion.py
Total Lines: 396
Pattern Category: AI Agent Design Pattern #126

---

Generated from actual implementation code
For educational and reference purposes