aishastr.live

Agentic Architecture Pattern: Graph Based Knowledge

Written by

admin

in


    """
Pattern 68: Graph-Based Knowledge Pattern

Agents store knowledge as a graph with nodes (entities) and edges (relationships).
Enables path finding, traversal, and graph-based reasoning.
"""

from typing import Dict, Any, List, Optional, Set, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class NodeType(Enum):
    """Type of knowledge graph node."""
    ENTITY = "entity"
    CONCEPT = "concept"
    EVENT = "event"
    ATTRIBUTE = "attribute"


@dataclass
class GraphNode:
    """Node in knowledge graph."""
    node_id: str
    node_type: NodeType
    label: str
    properties: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)


@dataclass
class GraphEdge:
    """Edge in knowledge graph."""
    edge_id: str
    source_id: str
    target_id: str
    relationship: str
    weight: float = 1.0
    properties: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)


@dataclass
class GraphPath:
    """Path through knowledge graph."""
    nodes: List[str]
    edges: List[str]
    total_weight: float
    path_length: int


class KnowledgeGraph:
    """Graph-based knowledge storage with traversal."""
    
    def __init__(self):
        self.nodes: Dict[str, GraphNode] = {}
        self.edges: Dict[str, GraphEdge] = {}
        self.adjacency: Dict[str, Set[str]] = {}  # node_id -> {edge_ids}
    
    def add_node(self, node: GraphNode):
        """Add node to graph."""
        self.nodes[node.node_id] = node
        if node.node_id not in self.adjacency:
            self.adjacency[node.node_id] = set()
    
    def add_edge(self, edge: GraphEdge):
        """Add edge to graph."""
        self.edges[edge.edge_id] = edge
        
        # Update adjacency
        if edge.source_id not in self.adjacency:
            self.adjacency[edge.source_id] = set()
        self.adjacency[edge.source_id].add(edge.edge_id)
    
    def get_neighbors(self, node_id: str) -> List[Tuple[GraphNode, GraphEdge]]:
        """Get neighboring nodes and connecting edges."""
        neighbors = []
        
        if node_id not in self.adjacency:
            return neighbors
        
        for edge_id in self.adjacency[node_id]:
            edge = self.edges.get(edge_id)
            if edge:
                target_node = self.nodes.get(edge.target_id)
                if target_node:
                    neighbors.append((target_node, edge))
        
        return neighbors
    
    def find_path(self, start_id: str, end_id: str, 
                  max_depth: int = 5) -> Optional[GraphPath]:
        """Find path between two nodes using BFS."""
        if start_id not in self.nodes or end_id not in self.nodes:
            return None
        
        # BFS
        queue = [(start_id, [start_id], [], 0.0)]
        visited = {start_id}
        
        while queue:
            current_id, path_nodes, path_edges, total_weight = queue.pop(0)
            
            if current_id == end_id:
                return GraphPath(
                    nodes=path_nodes,
                    edges=path_edges,
                    total_weight=total_weight,
                    path_length=len(path_nodes) - 1
                )
            
            if len(path_nodes) >= max_depth:
                continue
            
            # Explore neighbors
            for neighbor, edge in self.get_neighbors(current_id):
                if neighbor.node_id not in visited:
                    visited.add(neighbor.node_id)
                    queue.append((
                        neighbor.node_id,
                        path_nodes + [neighbor.node_id],
                        path_edges + [edge.edge_id],
                        total_weight + edge.weight
                    ))
        
        return None
    
    def find_nodes_by_type(self, node_type: NodeType) -> List[GraphNode]:
        """Find all nodes of a specific type."""
        return [node for node in self.nodes.values() 
                if node.node_type == node_type]
    
    def find_related(self, node_id: str, relationship: str) -> List[GraphNode]:
        """Find nodes related by specific relationship."""
        related = []
        
        if node_id not in self.adjacency:
            return related
        
        for edge_id in self.adjacency[node_id]:
            edge = self.edges.get(edge_id)
            if edge and edge.relationship == relationship:
                target = self.nodes.get(edge.target_id)
                if target:
                    related.append(target)
        
        return related
    
    def get_subgraph(self, node_id: str, depth: int = 2) -> Tuple[Set[str], Set[str]]:
        """Get subgraph around a node."""
        nodes = {node_id}
        edges = set()
        
        queue = [(node_id, 0)]
        visited = {node_id}
        
        while queue:
            current_id, current_depth = queue.pop(0)
            
            if current_depth >= depth:
                continue
            
            for neighbor, edge in self.get_neighbors(current_id):
                edges.add(edge.edge_id)
                if neighbor.node_id not in visited:
                    visited.add(neighbor.node_id)
                    nodes.add(neighbor.node_id)
                    queue.append((neighbor.node_id, current_depth + 1))
        
        return nodes, edges


class GraphKnowledgeAgent:
    """Agent with graph-based knowledge representation."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.graph = KnowledgeGraph()
    
    def learn_entity(self, entity_id: str, label: str, 
                     properties: Optional[Dict[str, Any]] = None):
        """Learn about an entity."""
        node = GraphNode(
            node_id=entity_id,
            node_type=NodeType.ENTITY,
            label=label,
            properties=properties or {}
        )
        self.graph.add_node(node)
        print(f"   šŸ”· {self.name} learned entity: {label}")
    
    def learn_relationship(self, source_id: str, target_id: str, 
                          relationship: str, weight: float = 1.0):
        """Learn a relationship between entities."""
        edge_id = f"{source_id}_{relationship}_{target_id}"
        edge = GraphEdge(
            edge_id=edge_id,
            source_id=source_id,
            target_id=target_id,
            relationship=relationship,
            weight=weight
        )
        self.graph.add_edge(edge)
        print(f"   šŸ”— {self.name} learned: {source_id} --{relationship}-> {target_id}")
    
    def query_relationship(self, entity_id: str, relationship: str) -> str:
        """Query relationships from an entity."""
        related = self.graph.find_related(entity_id, relationship)
        
        if not related:
            return f"No {relationship} relationships found for {entity_id}."
        
        entity = self.graph.nodes.get(entity_id)
        entity_label = entity.label if entity else entity_id
        
        related_labels = [node.label for node in related]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} with graph-based knowledge."),
            ("human", f"Entity: {entity_label}\n"
                     f"Relationship: {relationship}\n"
                     f"Related entities: {', '.join(related_labels)}\n\n"
                     f"Provide a natural language summary of this relationship. (1-2 sentences)")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content)[:200]
    
    def find_connection(self, entity1_id: str, entity2_id: str) -> str:
        """Find how two entities are connected."""
        path = self.graph.find_path(entity1_id, entity2_id)
        
        if not path:
            return f"No connection found between {entity1_id} and {entity2_id}."
        
        # Build path description
        path_desc = []
        for i in range(len(path.nodes) - 1):
            node1 = self.graph.nodes.get(path.nodes[i])
            node2 = self.graph.nodes.get(path.nodes[i + 1])
            edge = self.graph.edges.get(path.edges[i])
            
            if node1 and node2 and edge:
                path_desc.append(f"{node1.label} --{edge.relationship}-> {node2.label}")
        
        return f"Path: {' ā†’ '.join(path_desc)} (length: {path.path_length}, weight: {path.total_weight:.1f})"


class GraphBasedKnowledgePattern:
    """
    Graph-Based Knowledge Pattern - Knowledge as graph structures.
    
    Agents represent knowledge as nodes and edges, enabling graph
    traversal, path finding, and relationship-based reasoning.
    """
    
    def __init__(self):
        self.agents: Dict[str, GraphKnowledgeAgent] = {}
    
    def register_agent(self, agent: GraphKnowledgeAgent):
        """Register agent with graph knowledge."""
        self.agents[agent.name] = agent
    
    def run(self, domain: str) -> Dict[str, Any]:
        """Execute graph knowledge pattern."""
        total_nodes = sum(len(agent.graph.nodes) for agent in self.agents.values())
        total_edges = sum(len(agent.graph.edges) for agent in self.agents.values())
        
        return {
            "pattern": "GraphBasedKnowledge",
            "domain": domain,
            "total_agents": len(self.agents),
            "total_nodes": total_nodes,
            "total_edges": total_edges
        }


# Example usage
if __name__ == "__main__":
    print("=== Graph-Based Knowledge Pattern Demo ===\n")
    
    # Create pattern
    pattern = GraphBasedKnowledgePattern()
    
    # Create agent with graph knowledge
    print("1. Creating agent with graph knowledge...")
    agent = GraphKnowledgeAgent("Scholar")
    pattern.register_agent(agent)
    
    # Build knowledge graph about a company
    print("\n2. Building knowledge graph...")
    
    # Entities
    agent.learn_entity("company", "TechCorp", {"type": "company", "founded": 2010})
    agent.learn_entity("ceo", "Alice Johnson", {"role": "CEO"})
    agent.learn_entity("cto", "Bob Smith", {"role": "CTO"})
    agent.learn_entity("product1", "CloudApp", {"category": "SaaS"})
    agent.learn_entity("product2", "DataPlatform", {"category": "Analytics"})
    agent.learn_entity("customer1", "Acme Corp", {"type": "enterprise"})
    agent.learn_entity("tech1", "Python", {"type": "language"})
    agent.learn_entity("tech2", "Kubernetes", {"type": "platform"})
    
    # Relationships
    print("\n3. Adding relationships...")
    agent.learn_relationship("ceo", "company", "leads", 1.0)
    agent.learn_relationship("cto", "company", "works_for", 1.0)
    agent.learn_relationship("cto", "ceo", "reports_to", 1.0)
    agent.learn_relationship("company", "product1", "develops", 1.0)
    agent.learn_relationship("company", "product2", "develops", 1.0)
    agent.learn_relationship("customer1", "product1", "uses", 1.0)
    agent.learn_relationship("product1", "tech1", "built_with", 1.0)
    agent.learn_relationship("product2", "tech1", "built_with", 1.0)
    agent.learn_relationship("product2", "tech2", "deployed_on", 1.0)
    
    # Query relationships
    print("\n4. Querying relationships...")
    summary = agent.query_relationship("company", "develops")
    print(f"   {summary}")
    
    summary = agent.query_relationship("product1", "built_with")
    print(f"   {summary}")
    
    # Find connections
    print("\n5. Finding connections between entities...")
    connections = [
        ("customer1", "tech1"),
        ("cto", "customer1"),
        ("tech2", "ceo")
    ]
    
    for entity1, entity2 in connections:
        print(f"\n   Connection: {entity1}  {entity2}")
        result = agent.find_connection(entity1, entity2)
        print(f"   {result}")
    
    # Explore subgraph
    print("\n6. Exploring subgraph around 'company'...")
    nodes, edges = agent.graph.get_subgraph("company", depth=2)
    print(f"   Nodes in subgraph: {len(nodes)}")
    print(f"   Edges in subgraph: {len(edges)}")
    
    node_labels = [agent.graph.nodes[nid].label for nid in nodes if nid in agent.graph.nodes]
    print(f"   Entities: {', '.join(node_labels)}")
    
    # Find by type
    print("\n7. Finding entities by type...")
    entities = agent.graph.find_nodes_by_type(NodeType.ENTITY)
    print(f"   Total entities: {len(entities)}")
    for entity in entities[:5]:
        print(f"   - {entity.label} ({entity.properties.get('type', 'N/A')})")
    
    # Statistics
    print("\n8. Graph Statistics:")
    summary = pattern.run("Company Knowledge")
    print(f"   Total Nodes: {summary['total_nodes']}")
    print(f"   Total Edges: {summary['total_edges']}")
    print(f"   Avg connections per node: {summary['total_edges'] / max(1, summary['total_nodes']):.1f}")
    
    # Analyze graph structure
    print("\n9. Graph Structure Analysis:")
    node_degrees = {}
    for node_id in agent.graph.nodes:
        degree = len(agent.graph.adjacency.get(node_id, set()))
        node_degrees[node_id] = degree
    
    if node_degrees:
        max_degree = max(node_degrees.values())
        central_nodes = [nid for nid, deg in node_degrees.items() if deg == max_degree]
        print(f"   Max node degree: {max_degree}")
        print(f"   Most connected: {[agent.graph.nodes[nid].label for nid in central_nodes]}")
    
    print("\nāœ“ Graph-based knowledge pattern enabled relationship reasoning!")

Agentic Architecture Pattern: Graph Based Knowledge

Pattern 68: 68 Graph Based Knowledge – Line-by-Line Explanation

Overview

Pattern 68: Graph-Based Knowledge Pattern

Agents store knowledge as a graph with nodes (entities) and edges (relationships). Enables path finding, traversal, and graph-based reasoning.

Code Structure Analysis

File Information

  • Pattern Number: 68
  • Pattern Name: 68 Graph Based Knowledge
  • File Path: patterns\memory_knowledge\68_graph_based_knowledge.py
  • Total Lines: 379

Detailed Code Explanation

Module Docstring

"""
Pattern 68: Graph-Based Knowledge Pattern

Agents store knowledge as a graph with nodes (entities) and edges (relationships).
Enables path finding, traversal, and graph-based reasoning.
"""

Explanation: This module docstring describes the purpose and key features of the 68 Graph Based Knowledge pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set, Tuple: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

NodeType (Line 16)

class NodeType(Enum):
    ENTITY
    CONCEPT
    EVENT
    ATTRIBUTE

Purpose: Type of knowledge graph node.

Values: – ENTITY: Represents entity – CONCEPT: Represents concept – EVENT: Represents event – ATTRIBUTE: Represents attribute

Classes

NodeType (Line 16)

Description: Type of knowledge graph node.

GraphNode (Line 25)

Description: Node in knowledge graph.

GraphEdge (Line 35)

Description: Edge in knowledge graph.

GraphPath (Line 47)

Description: Path through knowledge graph.

KnowledgeGraph (Line 55)

Description: Graph-based knowledge storage with traversal.

Methods:

__init__ (Line 58)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

add_node (Line 63)
def add_node(self, node):
    """Add node to graph."""

Purpose: Add node to graph.

Parameters: – self: Instance reference – node: Parameter for node

add_edge (Line 69)
def add_edge(self, edge):
    """Add edge to graph."""

Purpose: Add edge to graph.

Parameters: – self: Instance reference – edge: Parameter for edge

get_neighbors (Line 78)
def get_neighbors(self, node_id):
    """Get neighboring nodes and connecting edges."""

Purpose: Get neighboring nodes and connecting edges.

Parameters: – self: Instance reference – node_id: Parameter for node_id

find_path (Line 94)
def find_path(self, start_id, end_id, max_depth):
    """Find path between two nodes using BFS."""

Purpose: Find path between two nodes using BFS.

Parameters: – self: Instance reference – start_id: Parameter for start_id – end_id: Parameter for end_id – max_depth: Parameter for max_depth

find_nodes_by_type (Line 131)
def find_nodes_by_type(self, node_type):
    """Find all nodes of a specific type."""

Purpose: Find all nodes of a specific type.

Parameters: – self: Instance reference – node_type: Parameter for node_type

find_related (Line 136)
def find_related(self, node_id, relationship):
    """Find nodes related by specific relationship."""

Purpose: Find nodes related by specific relationship.

Parameters: – self: Instance reference – node_id: Parameter for node_id – relationship: Parameter for relationship

get_subgraph (Line 152)
def get_subgraph(self, node_id, depth):
    """Get subgraph around a node."""

Purpose: Get subgraph around a node.

Parameters: – self: Instance reference – node_id: Parameter for node_id – depth: Parameter for depth

GraphKnowledgeAgent (Line 176)

Description: Agent with graph-based knowledge representation.

Methods:

__init__ (Line 179)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

learn_entity (Line 184)
def learn_entity(self, entity_id, label, properties):
    """Learn about an entity."""

Purpose: Learn about an entity.

Parameters: – self: Instance reference – entity_id: Parameter for entity_id – label: Parameter for label – properties: Parameter for properties

learn_relationship (Line 196)
def learn_relationship(self, source_id, target_id, relationship, weight):
    """Learn a relationship between entities."""

Purpose: Learn a relationship between entities.

Parameters: – self: Instance reference – source_id: Parameter for source_id – target_id: Parameter for target_id – relationship: Parameter for relationship – weight: Parameter for weight

query_relationship (Line 210)
def query_relationship(self, entity_id, relationship):
    """Query relationships from an entity."""

Purpose: Query relationships from an entity.

Parameters: – self: Instance reference – entity_id: Parameter for entity_id – relationship: Parameter for relationship

find_connection (Line 235)
def find_connection(self, entity1_id, entity2_id):
    """Find how two entities are connected."""

Purpose: Find how two entities are connected.

Parameters: – self: Instance reference – entity1_id: Parameter for entity1_id – entity2_id: Parameter for entity2_id

GraphBasedKnowledgePattern (Line 255)

Description: Graph-Based Knowledge Pattern – Knowledge as graph structures.

Agents represent knowledge as nodes and edges, enabling graph traversal, path finding, and relationship-based reasoning.

Methods:

__init__ (Line 263)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 266)
def register_agent(self, agent):
    """Register agent with graph knowledge."""

Purpose: Register agent with graph knowledge.

Parameters: – self: Instance reference – agent: Parameter for agent

run (Line 270)
def run(self, domain):
    """Execute graph knowledge pattern."""

Purpose: Execute graph knowledge pattern.

Parameters: – self: Instance reference – domain: Parameter for domain

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = NodeType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

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

More posts