aishastr.live

Agentic Design Pattern: Knowledge Graph Integration

Written by

admin

in


    """
Pattern 123: Knowledge Graph Integration

Description:
    Uses structured knowledge graphs for reasoning with graph traversal,
    pattern matching, and inference capabilities.

Components:
    - Knowledge graph
    - Graph traversal
    - Pattern matching
    - Inference engine

Use Cases:
    - Complex reasoning
    - Knowledge-intensive tasks
    - Relationship discovery

LangChain Implementation:
    Integrates knowledge graphs with LLM reasoning for enhanced understanding.
"""

import os
from typing import List, Dict, Any, Set, Tuple
from dataclasses import dataclass
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


@dataclass
class Node:
    """Represents a node in the knowledge graph."""
    id: str
    type: str
    properties: Dict[str, Any]


@dataclass
class Edge:
    """Represents an edge in the knowledge graph."""
    source: str
    target: str
    relationship: str
    properties: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.properties is None:
            self.properties = {}


class KnowledgeGraph:
    """Knowledge graph for structured knowledge representation."""
    
    def __init__(self):
        self.nodes: Dict[str, Node] = {}
        self.edges: List[Edge] = []
        
    def add_node(self, node: Node):
        """Add a node to the graph."""
        self.nodes[node.id] = node
        
    def add_edge(self, edge: Edge):
        """Add an edge to the graph."""
        self.edges.append(edge)
        
    def get_neighbors(self, node_id: str, relationship: str = None) -> List[Tuple[Node, str]]:
        """Get neighboring nodes."""
        neighbors = []
        for edge in self.edges:
            if edge.source == node_id:
                if relationship is None or edge.relationship == relationship:
                    if edge.target in self.nodes:
                        neighbors.append((self.nodes[edge.target], edge.relationship))
        return neighbors
    
    def find_path(self, start_id: str, end_id: str, max_depth: int = 5) -> List[str]:
        """Find path between two nodes."""
        visited = set()
        queue = [(start_id, [start_id])]
        
        while queue:
            current, path = queue.pop(0)
            
            if current == end_id:
                return path
            
            if len(path) >= max_depth:
                continue
            
            if current in visited:
                continue
            
            visited.add(current)
            
            for neighbor, _ in self.get_neighbors(current):
                if neighbor.id not in visited:
                    queue.append((neighbor.id, path + [neighbor.id]))
        
        return []
    
    def query_subgraph(self, node_id: str, depth: int = 2) -> Dict[str, Any]:
        """Extract subgraph around a node."""
        visited = set()
        subgraph_nodes = {}
        subgraph_edges = []
        
        def explore(nid, d):
            if d > depth or nid in visited:
                return
            visited.add(nid)
            
            if nid in self.nodes:
                subgraph_nodes[nid] = self.nodes[nid]
                
                for neighbor, relationship in self.get_neighbors(nid):
                    edge = Edge(nid, neighbor.id, relationship)
                    subgraph_edges.append(edge)
                    explore(neighbor.id, d + 1)
        
        explore(node_id, 0)
        
        return {
            "nodes": subgraph_nodes,
            "edges": subgraph_edges
        }


class KnowledgeGraphAgent:
    """Agent that uses knowledge graph for reasoning."""
    
    def __init__(self, kg: KnowledgeGraph, model_name: str = "gpt-4"):
        self.kg = kg
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
    def answer_question(self, question: str) -> str:
        """Answer question using knowledge graph."""
        # Extract relevant entities from question
        entity_extraction_prompt = ChatPromptTemplate.from_messages([
            ("system", "Extract key entities from this question."),
            ("user", "Question: {question}\n\nEntities (comma-separated):")
        ])
        
        chain = entity_extraction_prompt | self.llm | StrOutputParser()
        entities_str = chain.invoke({"question": question})
        entities = [e.strip() for e in entities_str.split(",")]
        
        # Find relevant nodes
        relevant_nodes = []
        for entity in entities:
            for node_id, node in self.kg.nodes.items():
                if entity.lower() in node.properties.get("name", "").lower():
                    relevant_nodes.append(node)
        
        # Extract relevant subgraph
        kg_context = ""
        for node in relevant_nodes[:3]:  # Limit to top 3
            subgraph = self.kg.query_subgraph(node.id, depth=2)
            
            # Format subgraph info
            node_info = f"\nEntity: {node.properties.get('name', node.id)}\n"
            for edge in subgraph['edges']:
                if edge.source in subgraph['nodes'] and edge.target in subgraph['nodes']:
                    source_name = subgraph['nodes'][edge.source].properties.get('name', edge.source)
                    target_name = subgraph['nodes'][edge.target].properties.get('name', edge.target)
                    node_info += f"  {source_name} --[{edge.relationship}]--> {target_name}\n"
            
            kg_context += node_info
        
        # Answer using knowledge graph context
        qa_prompt = ChatPromptTemplate.from_messages([
            ("system", "Answer the question using the knowledge graph information."),
            ("user", """Knowledge Graph Context:
{kg_context}

Question: {question}

Answer:""")
        ])
        
        chain = qa_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "kg_context": kg_context or "No relevant information found in knowledge graph",
            "question": question
        })
    
    def infer_relationships(self, entity1: str, entity2: str) -> str:
        """Infer relationships between entities."""
        # Find nodes
        node1 = None
        node2 = None
        
        for node_id, node in self.kg.nodes.items():
            if entity1.lower() in node.properties.get("name", "").lower():
                node1 = node
            if entity2.lower() in node.properties.get("name", "").lower():
                node2 = node
        
        if not node1 or not node2:
            return f"Could not find entities in knowledge graph"
        
        # Find path
        path = self.kg.find_path(node1.id, node2.id)
        
        if not path:
            return f"No direct relationship found between {entity1} and {entity2}"
        
        # Construct relationship chain
        relationships = []
        for i in range(len(path) - 1):
            for edge in self.kg.edges:
                if edge.source == path[i] and edge.target == path[i+1]:
                    source_name = self.kg.nodes[edge.source].properties.get('name', edge.source)
                    target_name = self.kg.nodes[edge.target].properties.get('name', edge.target)
                    relationships.append(f"{source_name} --[{edge.relationship}]--> {target_name}")
        
        # Generate inference
        inference_prompt = ChatPromptTemplate.from_messages([
            ("system", "Explain the relationship between entities based on this knowledge graph path."),
            ("user", """Entity 1: {entity1}
Entity 2: {entity2}

Relationship Chain:
{chain}

Explain the relationship:""")
        ])
        
        chain_str = "\n".join(relationships)
        chain = inference_prompt | self.llm | StrOutputParser()
        
        return chain.invoke({
            "entity1": entity1,
            "entity2": entity2,
            "chain": chain_str
        })
    
    def discover_patterns(self) -> str:
        """Discover patterns in the knowledge graph."""
        # Analyze graph structure
        structure_info = f"Total nodes: {len(self.kg.nodes)}\n"
        structure_info += f"Total edges: {len(self.kg.edges)}\n\n"
        
        # Common relationships
        relationship_counts = {}
        for edge in self.kg.edges:
            relationship_counts[edge.relationship] = relationship_counts.get(edge.relationship, 0) + 1
        
        structure_info += "Relationship types:\n"
        for rel, count in sorted(relationship_counts.items(), key=lambda x: x[1], reverse=True):
            structure_info += f"  {rel}: {count}\n"
        
        pattern_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze this knowledge graph structure and identify interesting patterns."),
            ("user", """Graph Structure:
{structure}

Identify:
1. Central entities (hubs)
2. Common patterns
3. Interesting insights
4. Potential gaps""")
        ])
        
        chain = pattern_prompt | self.llm | StrOutputParser()
        return chain.invoke({"structure": structure_info})


def demonstrate_knowledge_graph_integration():
    """Demonstrate knowledge graph integration pattern."""
    print("=== Knowledge Graph Integration Pattern ===\n")
    
    # Create knowledge graph
    print("1. Building Knowledge Graph")
    print("-" * 50)
    
    kg = KnowledgeGraph()
    
    # Add nodes (people, companies, technologies)
    kg.add_node(Node("person1", "Person", {"name": "Alice", "role": "CEO"}))
    kg.add_node(Node("person2", "Person", {"name": "Bob", "role": "CTO"}))
    kg.add_node(Node("person3", "Person", {"name": "Charlie", "role": "Engineer"}))
    
    kg.add_node(Node("company1", "Company", {"name": "TechCorp", "industry": "Software"}))
    kg.add_node(Node("company2", "Company", {"name": "DataInc", "industry": "Analytics"}))
    
    kg.add_node(Node("tech1", "Technology", {"name": "Python", "type": "Language"}))
    kg.add_node(Node("tech2", "Technology", {"name": "Machine Learning", "type": "Field"}))
    kg.add_node(Node("tech3", "Technology", {"name": "TensorFlow", "type": "Framework"}))
    
    # Add edges (relationships)
    kg.add_edge(Edge("person1", "company1", "WORKS_AT"))
    kg.add_edge(Edge("person2", "company1", "WORKS_AT"))
    kg.add_edge(Edge("person3", "company2", "WORKS_AT"))
    
    kg.add_edge(Edge("person1", "person2", "MANAGES"))
    kg.add_edge(Edge("person2", "person3", "MENTORS"))
    
    kg.add_edge(Edge("person2", "tech1", "EXPERT_IN"))
    kg.add_edge(Edge("person2", "tech2", "EXPERT_IN"))
    kg.add_edge(Edge("person3", "tech1", "USES"))
    kg.add_edge(Edge("person3", "tech3", "USES"))
    
    kg.add_edge(Edge("tech3", "tech1", "IMPLEMENTED_IN"))
    kg.add_edge(Edge("tech3", "tech2", "USED_FOR"))
    
    kg.add_edge(Edge("company1", "tech2", "SPECIALIZES_IN"))
    kg.add_edge(Edge("company2", "tech2", "USES"))
    
    print(f"✓ Created knowledge graph with {len(kg.nodes)} nodes and {len(kg.edges)} edges")
    print()
    
    # Create agent
    agent = KnowledgeGraphAgent(kg)
    
    # Question answering
    print("2. Question Answering with Knowledge Graph")
    print("-" * 50)
    
    questions = [
        "Who works at TechCorp?",
        "What technologies does Bob know?",
        "Which companies use Machine Learning?"
    ]
    
    for question in questions:
        print(f"\nQ: {question}")
        answer = agent.answer_question(question)
        print(f"A: {answer}")
    
    # Relationship inference
    print("\n\n3. Inferring Relationships")
    print("-" * 50)
    
    print("\nInferring relationship between Alice and Charlie:")
    inference = agent.infer_relationships("Alice", "Charlie")
    print(inference)
    
    print("\nInferring relationship between Bob and TensorFlow:")
    inference = agent.infer_relationships("Bob", "TensorFlow")
    print(inference)
    
    # Pattern discovery
    print("\n\n4. Discovering Patterns in Knowledge Graph")
    print("-" * 50)
    patterns = agent.discover_patterns()
    print(patterns)
    
    print("\n=== Summary ===")
    print("Knowledge graph integration demonstrated with:")
    print("- Graph construction (nodes and edges)")
    print("- Graph traversal and querying")
    print("- Question answering with graph context")
    print("- Relationship inference")
    print("- Pattern discovery")
    print("- LLM-enhanced graph reasoning")


if __name__ == "__main__":
    demonstrate_knowledge_graph_integration()

Agentic Design Pattern: Knowledge Graph Integration

Pattern 123: Knowledge Graph Integration – Line-by-Line Explanation

Overview

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

Pattern Description

Pattern 123: Knowledge Graph Integration

Description: Uses structured knowledge graphs for reasoning with graph traversal, pattern matching, and inference capabilities.

Components: – Knowledge graph – Graph traversal – Pattern matching – Inference engine

Use Cases: – Complex reasoning – Knowledge-intensive tasks – Relationship discovery

LangChain Implementation: Integrates knowledge graphs with LLM reasoning for enhanced understanding.

File Statistics

  • Total Lines: 365
  • Classes: 4
  • Functions: 1
  • Constants: 0

Module Documentation

"""
Pattern 123: Knowledge Graph Integration

Description:
    Uses structured knowledge graphs for reasoning with graph traversal,
    pattern matching, and inference capabilities.

Components:
    - Knowledge graph
    - Graph traversal
    - Pattern matching
    - Inference engine

Use Cases:
    - Complex reasoning
    - Knowledge-intensive tasks
    - Relationship discovery

LangChain Implementation:
    Integrates knowledge graphs with LLM reasoning for enhanced understanding.
"""

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

Import Statements

Standard Library Imports

import dataclasses.dataclass
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Set
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: 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.Set: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: 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: Node

Lines 36-40

Purpose:

Represents a node in the knowledge graph.

Class: Edge

Lines 44-53

Purpose:

Represents an edge in the knowledge graph.

Methods:

__post_init__(self)

Lines 51-53

Purpose: Post Init operation.

Parameters:

Class: KnowledgeGraph

Lines 56-130

Purpose:

Knowledge graph for structured knowledge representation.

Methods:

__init__(self)

Lines 59-61

Purpose: Initialize the class instance with required parameters.

Parameters:

add_node(self, node)

Lines 63-65

Purpose:

Add a node to the graph.

Parameters: – node: Node

add_edge(self, edge)

Lines 67-69

Purpose:

Add an edge to the graph.

Parameters: – edge: Edge

get_neighbors(self, node_id, relationship)

Lines 71-79

Purpose:

Get neighboring nodes.

Parameters: – node_id: Node id – relationship: Relationship

Returns: <ast.Subscript object at 0x000001A06D6EB9D0>

find_path(self, start_id, end_id, max_depth)

Lines 81-104

Purpose:

Find path between two nodes.

Parameters: – start_id: Start id – end_id: End id – max_depth: Max depth

Returns: <ast.Subscript object at 0x000001A06D63BD90>

query_subgraph(self, node_id, depth)

Lines 106-130

Purpose:

Extract subgraph around a node.

Parameters: – node_id: Node id – depth: Depth

Returns: <ast.Subscript object at 0x000001A06D646E50>

Class: KnowledgeGraphAgent

Lines 133-270

Purpose:

Agent that uses knowledge graph for reasoning.

Methods:

__init__(self, kg, model_name)

Lines 136-138

Purpose: Initialize the class instance with required parameters.

Parameters: – kg: Kg – model_name: Language model instance

answer_question(self, question)

Lines 140-189

Purpose:

Answer question using knowledge graph.

Parameters: – question: Question to answer

Returns: str

infer_relationships(self, entity1, entity2)

Lines 191-240

Purpose:

Infer relationships between entities.

Parameters: – entity1: Entity1 – entity2: Entity2

Returns: str

discover_patterns(self)

Lines 242-270

Purpose:

Discover patterns in the knowledge graph.

Parameters:

Returns: str

Functions

demonstrate_knowledge_graph_integration()

Lines 273-360

Purpose:

Demonstrate knowledge graph integration pattern.

Key Concepts Summary

Pattern Overview

The Knowledge Graph Integration 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 122
  • Pattern 124

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 Graph Integration 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: 123_knowledge_graph_integration.py
Total Lines: 365
Pattern Category: AI Agent Design Pattern #123

Generated from actual implementation code
For educational and reference purposes

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts