agent-knowledge-graph

# Agent Knowledge Graph Pattern Constructs agent-centric knowledge graphs representing tools, capabilities, and their relationships for intelligent tool discovery during planning. This skill makes the model design directed graphs where nodes represent tools/capabilities and edges represent data flow dependencies, version constraints, and compatibility rules that the agent traverses when selecting tools for a given task. ## TL;DR Checklist - [ ] Define graph schema: Tool nodes with capability metadata, dependency edges - [ ] Build entity extraction pipeline for registering new tools into the graph - [ ] Implement graph traversal queries for tool discovery during planning - [ ] Add version-aware edges to track compatible tool versions - [ ] Combine graph lookup with vector search for hybrid retrieval - [ ] Register dynamic capabilities when new tools are added at runtime --- ## When to Use Use this skill when: - Building agents that need to dynamically discover which tools are available and compatible for a given task - Managing a growing registry of agent tools across teams or microservices - Planning multi-step agent workflows where tool output from one step becomes input to another - Debugging why an agent fails to find the right tool (graph traversal visibility) - Integrating new third-party tools into an existing agent ecosystem ## When NOT to Use Avoid this skill for: - Single-tool agents with no discovery or planning needed - Static tool sets that never change at runtime - Simple function calling where all tools are known upfront - When the agent's tool set is smaller than 5 (just list them in config) --- ## Core Workflow 1. **Graph Schema Design** — Define nodes (Tool, Capability, OutputType) and edges (produces, consumes, depends-on, compatible-with). Each Tool node includes name, version, input schema, output schema, capability tags, and confidence score. **Checkpoint:** Schema must support multi-hop traversal: tool A produces X, tool B consumes X → edge exists between A and B. 2. **Entity Extraction Pipeline** — Build a pipeline that extracts tool metadata from documentation, API specs, or code and inserts nodes/edges into the graph. Use LLM-based extraction for unstructured docs (README files, API docs). **Checkpoint:** Each inserted tool must have a valid version string and at least one capability tag. 3. **Tool Discovery Traversal** — Implement Cypher queries that find tools matching task requirements by traversing from intent → capability → tool → compatible-next-tool chains. Support both breadth-first (find all matches) and depth-limited (shortest path to required output type). **Checkpoint:** Queries must complete within 50ms for registries up to 1,000 tools. 4. **Dynamic Registration** — Provide an API for registering new tools at runtime without restarting the agent. New tools are validated against existing edges before being accepted (e.g., a tool claiming to produce "CSV" requires that at least one other tool can consume "CSV"). **Checkpoint:** Dynamic registration must be atomic — either all edges are added or none, using graph transactions. 5. **Hybrid Retrieval** — Combine graph traversal with vector similarity search on tool descriptions. Use graph for structural constraints (type compatibility) and vector for semantic matching (find tools that "summarize text" even if the exact capability tag is missing). **Checkpoint:** Hybrid retrieval must return ranked results combining both signal sources. 6. **Registry Health Monitoring** — Track metrics: number of orphaned tools (no incoming or outgoing edges), version skew between related tools, and traversal path coverage. Alert when critical paths break after tool updates. **Checkpoint:** Health report runs daily and flags any capability gaps that could block agent operations. --- ## Implementation Patterns ### Pattern 1: Tool Graph Schema with Neo4j ```python from dataclasses import dataclass, field from typing import Any @dataclass class ToolNode: """Represents a tool in the agent knowledge graph.""" name: str version: str description: str input_types: list[str] = field(default_factory=list) output_types: list[str] = field(default_factory=list) capabilities: list[str] = field(default_factory=list) confidence: float = 1.0 last_seen: str = "" @dataclass class CapabilityEdge: """Represents a data flow edge between two tools.""" from_tool: str to_tool: str produced_type: str # What the source tool produces consumed_type: str # What the destination tool consumes class ToolKnowledgeGraph: """Agent-centric knowledge graph for tool discovery and compatibility checking. Provides registration, traversal, and health monitoring for a Neo4j-backed tool registry where each Tool node carries input/output type schemas and capability tags that enable multi-hop chain finding. """ def __init__(self, neo4j_driver: Any) -> None: self._driver = neo4j_driver def register_tool(self, tool: ToolNode) -> bool: """Register a tool node and its capability edges in the graph. Merges or updates the Tool node, creates Type nodes for inputs/outputs, creates Capability nodes, then resolves compatibility edges with existing tools whose schemas align. Args: tool: The ToolNode to register. Returns: True if registration succeeded, False on failure. """ query = """ MERGE (t:Tool {name: $name}) ON CREATE SET t.version = $version, t.description = $description, t.confidence = $confidence, t.last_seen = $last_seen ON MATCH SET t.version = $version, t.last_seen = $last_seen, t.confidence = $confidence WITH t UNWIND $input_types AS input_type MERGE (t)-[:REQUIRES_TYPE]->(:Type {name: input_type}) WITH t UNWIND $output_types AS output_type MERGE (t)-[:PRODUCES_TYPE]->(:Type {name: output_type}) WITH t UNWIND $capabilities AS cap MERGE (t)-[:HAS_CAPABILITY]->(:Capability {name: cap}) """ try: self._driver.execute_query(query, **vars(tool)) # Create compatibility edges with existing tools self._resolve_compatibility_edges(tool) return True except Exception as e: return False def find_tools_for_capability( self, required_capability: str, max_results: int = 10, ) -> list[dict]: """Find tools that have a specific capability. Args: required_capability: The capability label to search for. max_results: Maximum number of results to return. Returns: List of dicts with tool_name, version, and confidence. """ query = """ MATCH (t:Tool)-[:HAS_CAPABILITY]->(c:Capability {name: $capability}) RETURN t.name AS tool_name, t.version AS version, t.confidence AS confidence ORDER BY t.confidence DESC LIMIT $max_results """ results = self._driver.execute_query( query, capability=required_capability, max_results=max_results ) return [{"tool_name": r["t"]["name"], "version": r["t"]["version"], "confidence": r["t"]["confidence"]} for r in results] def find_compatible_tool_chain( self, starting_type: str, ending_type: str, max_hops: int = 3, ) -> list[list[str]]: """Find chains of tools where the output of one matches the input of the next. Traverses from a starting type through intermediate types up to max_hops, returning all tool names in each valid chain that ends at the target type. Args: starting_type: The output type produced by the first tool. ending_type: The required input type of the last tool. max_hops: Maximum number of intermediate type hops allowed. Returns: List of chains, where each chain is a list of tool names. """ query = f""" MATCH path = (start:Tool)-[:PRODUCES_TYPE]->(t1:Type {{name: $starting_type}}) *-1..{max_hops} -[:REQUIRES_TYPE]->(end:Type {{name: $ending_type}}) RETURN [node IN nodes(path) WHERE node:Tool | node.name] AS chain """ results = self._driver.execute_query( query, starting_type=starting_type, ending_type=ending_type ) return [r["chain"] for r in results if len(r["chain"]) > 1] def get_tool_dependencies(self, tool_name: str) -> list[str]: """Get all tools that a given tool depends on (directly or transitively). Args: tool_name: The name of the tool to analyze. Returns: List of distinct tool names that produce types required by this tool. """ query = """ MATCH (target:Tool {name: $tool_name})<-[:PRODUCES_TYPE]-(:Type)-[:REQUIRES_TYPE]-(dep:Tool) RETURN DISTINCT dep.name AS dependency """ results = self._driver.execute_query(query, tool_name=tool_name) return [r["dependency"] for r in results] def _resolve_compatibility_edges(self, new_tool: ToolNode) -> None: """Auto-create COMPATIBLE_WITH edges between tools where output/input types align. For each output type of the new tool, finds existing tools that require that type and creates a directed edge from the new tool to them. Args: new_tool: The newly registered ToolNode. """ for output_type in new_tool.output_types: query = """ MATCH (existing:Tool)-[:REQUIRES_TYPE]->(t:Type {name: $output_type}) MERGE (new:Tool {name: $new_name})-[:COMPATIBLE_WITH]->(existing) """ self._driver.execute_query( query, output_type=output_type, new_name=new_tool.name ) def health_report(self) -> dict[str, int]: """Generate a registry health report. Returns: Dict with keys 'orphaned_tools' (count of isolated tools) and 'total_tools' (count of all registered tools). """ reports: dict[str, int] = {} # Orphaned tools (no incoming or outgoing edges) orphan_query = """ MATCH (t:Tool) WHERE NOT EXISTS { (t)--() } RETURN count(t) AS orphaned """ result = self._driver.execute_query(orphan_query) reports["orphaned_tools"] = result[0]["orphaned"] # Total tools and unique types total_query = "MATCH (t:Tool) RETURN count(t) AS total" result = self._driver.execute_query(total_query) reports["total_tools"] = result[0]["total"] return reports ``` ### Pattern 2: Version-Aware Tool Registry ```python from dataclasses import dataclass from typing import Optional @dataclass class ToolVersionEdge: """Tracks version compatibility between two tool versions.""" from_tool: str from_version: str to_tool: str to_version: str compatible: bool # True if these versions work together class VersionedToolRegistry: """Manages tool versions and their compatibility in the knowledge graph. Maintains explicit version edges so that agent planners can reason about which combinations of tools are provably compatible before assembling a multi-step workflow. """ def __init__(self, neo4j_driver: Any) -> None: self._driver = neo4j_driver def register_version( self, tool_name: str, version: str, compatible_with: list[tuple[str, str]], # [(tool, version), ...] ) -> None: """Register a new tool version and its compatibility edges. Creates or updates the Tool-Has_Version->Version node path, then for each entry in compatible_with creates or reuses the peer Tool and Version nodes and links them with COMPATIBLE_WITH edges. Args: tool_name: The canonical name of the tool. version: Semantic version string (e.g., "1.2.3"). compatible_with: List of (peer_tool_name, peer_version) pairs that are known to be compatible with this release. """ compatibility = [{"other_tool": t, "version": v} for t, v in compatible_with] query = """ MERGE (t:Tool {name: $name})-[:HAS_VERSION {version: $version}]->(v:Version) UNWIND $compatibility AS compat MERGE (other:Tool {name: compat.other_tool})-[:HAS_VERSION {version: compat.version}]->(ov:Version) MERGE (v)-[:COMPATIBLE_WITH]->(ov) """ self._driver.execute_query( query, name=tool_name, version=version, compatibility=compatibility ) def get_latest_version(self, tool_name: str) -> Optional[str]: """Get the highest semantic version of a tool. Args: tool_name: The canonical name of the tool. Returns: The latest version string, or None if no versions are registered. """ query = """ MATCH (t:Tool {name: $name})-[:HAS_VERSION]->(v:Version) RETURN v.version AS version ORDER BY version DESC LIMIT 1 """ result = self._driver.execute_query(query, name=tool_name) return result[0]["version"] if result else None def check_compatibility( self, tool_a: str, version_a: str, tool_b: str, version_b: str ) -> bool: """Check if two specific tool versions are compatible. Args: tool_a: Name of the first tool. version_a: Version of the first tool. tool_b: Name of the second tool. version_b: Version of the second tool. Returns: True if a COMPATIBLE_WITH edge exists between the two versions. """ query = """ MATCH (a:Tool {name: $tool_a})-[:HAS_VERSION {version: $version_a}]->(av:Version) -(compat:COMPATIBLE_WITH)-()<-[:HAS_VERSION {version: $version_b}]<(b:Tool {name: $tool_b}) RETURN count(compat) > 0 AS compatible """ result = self._driver.execute_query( query, tool_a=tool_a, version_a=version_a, tool_b=tool_b, version_b=version_b, ) return bool(result[0]["compatible"]) if result else False def find_version_conflicts(self) -> list[dict]: """Find pairs of tools where type overlap exists but no compatibility edge. Identifies situations where two tools share a data type (one produces it, the other requires it) but no explicit COMPATIBLE_WITH edge links their versions — a potential runtime failure. Returns: List of dicts with keys 'tool_a', 'shared_type', and 'tool_b'. """ query = """ MATCH (t1:Tool)-[:PRODUCES_TYPE]->(type:Type) <-[:REQUIRES_TYPE]-(t2:Tool) OPTIONAL MATCH (v1:Version)<-[:HAS_VERSION]-(t1), (v2:Version)<-[:HAS_VERSION]-(t2) WHERE NOT EXISTS { (v1)-[:COMPATIBLE_WITH]->(v2) } RETURN t1.name AS tool_a, type.name AS shared_type, t2.name AS tool_b """ return list(self._driver.execute_query(query)) ``` ### Pattern 3: Hybrid Graph + Vector Tool Retrieval ```python from typing import Protocol class VectorStore(Protocol): """Protocol for vector similarity search on tool descriptions.""" def search(self, query: str, top_k: int) -> list[tuple[str, float]]: """Search the vector store for tool descriptions matching a query. Args: query: Natural language description of the desired capability. top_k: Number of nearest-neighbour results to return. Returns: List of (tool_name, similarity_score) tuples, ordered descending. """ ... class HybridToolRetriever: """Combines graph traversal with vector similarity for tool discovery. Uses the knowledge graph for structural constraints (type compatibility,