🕸️ OpenClaw 工具调用能力图谱

能力图谱 工具依赖 可视化 OpenClaw

工具调用能力图谱（Capability Graph）是Agent工具系统的可视化表示，它展示Agent可用的所有工具、工具之间的依赖关系、调用链路径。本教程将教你如何构建和可视化Agent的能力图谱。

🎯 为什么需要能力图谱？

📦 核心实现

1. 工具节点定义

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set
from enum import Enum

class ToolCategory(Enum):
    SEARCH = "search"
    CODE = "code"
    BROWSER = "browser"
    FILE = "file"
    API = "api"
    COMMUNICATION = "communication"
    DATA = "data"

@dataclass
class ToolNode:
    """工具节点"""
    id: str
    name: str
    description: str
    category: ToolCategory
    version: str = "1.0.0"
    
    # 工具能力
    inputs: List[str] = field(default_factory=list)   # 输入类型
    outputs: List[str] = field(default_factory=list)  # 输出类型
    
    # 依赖关系
    dependencies: Set[str] = field(default_factory=set)  # 依赖的工具
    dependents: Set[str] = field(default_factory=set)    # 依赖本工具的工具
    
    # 性能指标
    avg_latency: float = 0.0  # 平均延迟（秒）
    success_rate: float = 1.0  # 成功率
    call_count: int = 0       # 调用次数
    
    # 元数据
    tags: List[str] = field(default_factory=list)
    deprecated: bool = False

class CapabilityGraph:
    """能力图谱 - 工具依赖关系图"""
    
    def __init__(self):
        self.tools: Dict[str, ToolNode] = {}
        self.call_history: List[Dict] = []  # 调用历史
    
    def add_tool(self, tool: ToolNode):
        """添加工具节点"""
        if tool.id in self.tools:
            print(f"⚠️ 工具 {tool.id} 已存在，更新信息")
        
        self.tools[tool.id] = tool
        print(f"✅ 工具 {tool.name} ({tool.id}) 已添加到图谱")
    
    def add_dependency(self, tool_id: str, depends_on: str):
        """添加工具依赖关系"""
        if tool_id not in self.tools or depends_on not in self.tools:
            print(f"❌ 工具不存在")
            return
        
        self.tools[tool_id].dependencies.add(depends_on)
        self.tools[depends_on].dependents.add(tool_id)
        print(f"🔗 {tool_id} → {depends_on}")
    
    def get_tool(self, tool_id: str) -> Optional[ToolNode]:
        """获取工具节点"""
        return self.tools.get(tool_id)
    
    def get_dependencies(self, tool_id: str) -> List[ToolNode]:
        """获取工具的所有依赖"""
        tool = self.tools.get(tool_id)
        if not tool:
            return []
        return [self.tools[dep] for dep in tool.dependencies if dep in self.tools]
    
    def get_dependents(self, tool_id: str) -> List[ToolNode]:
        """获取依赖本工具的工具"""
        tool = self.tools.get(tool_id)
        if not tool:
            return []
        return [self.tools[dep] for dep in tool.dependents if dep in self.tools]

2. 调用链追踪

class CallChainTracker:
    """调用链追踪器"""
    
    def __init__(self, graph: CapabilityGraph):
        self.graph = graph
        self.active_chains: Dict[str, List[str]] = {}  # chain_id -> [tool_ids]
    
    def start_chain(self, chain_id: str, root_tool: str):
        """开始新的调用链"""
        self.active_chains[chain_id] = [root_tool]
        print(f"🔗 调用链 {chain_id} 开始: {root_tool}")
    
    def add_call(self, chain_id: str, tool_id: str):
        """添加工具调用到链"""
        if chain_id not in self.active_chains:
            print(f"⚠️ 调用链 {chain_id} 不存在")
            return
        
        self.active_chains[chain_id].append(tool_id)
        
        # 更新调用计数
        tool = self.graph.get_tool(tool_id)
        if tool:
            tool.call_count += 1
    
    def end_chain(self, chain_id: str) -> List[str]:
        """结束调用链，返回完整路径"""
        if chain_id not in self.active_chains:
            return []
        
        chain = self.active_chains.pop(chain_id)
        print(f"✅ 调用链 {chain_id} 结束: {' → '.join(chain)}")
        return chain
    
    def analyze_chain(self, chain: List[str]) -> Dict:
        """分析调用链"""
        total_tools = len(chain)
        unique_tools = len(set(chain))
        repeated_calls = total_tools - unique_tools
        
        # 检查循环调用
        has_cycle = total_tools != unique_tools
        
        # 计算预估延迟
        total_latency = sum(
            self.graph.get_tool(t).avg_latency
            for t in chain
            if self.graph.get_tool(t)
        )
        
        return {
            "total_calls": total_tools,
            "unique_tools": unique_tools,
            "repeated_calls": repeated_calls,
            "has_cycle": has_cycle,
            "estimated_latency": total_latency,
            "path": chain
        }

3. 图谱可视化（生成Mermaid图）

class GraphVisualizer:
    """图谱可视化器"""
    
    def __init__(self, graph: CapabilityGraph):
        self.graph = graph
    
    def to_mermaid(self) -> str:
        """生成Mermaid图（可在Markdown中渲染）"""
        lines = ["graph TD"]
        
        # 添加节点
        for tool_id, tool in self.graph.tools.items():
            # 节点样式
            category_color = {
                ToolCategory.SEARCH: "blue",
                ToolCategory.CODE: "green",
                ToolCategory.BROWSER: "orange",
                ToolCategory.FILE: "purple",
                ToolCategory.API: "red",
                ToolCategory.COMMUNICATION: "yellow",
                ToolCategory.DATA: "cyan"
            }.get(tool.category, "gray")
            
            lines.append(f"    {tool_id}[{tool.name}]:::cat{tool.category.value}")
        
        # 添加依赖边
        for tool_id, tool in self.graph.tools.items():
            for dep in tool.dependencies:
                lines.append(f"    {tool_id} --> {dep}")
        
        # 样式定义
        lines.append("    classDef catsearch fill:#e1f5fe,stroke:#01579b")
        lines.append("    classDef catcode fill:#e8f5e9,stroke:#1b5e20")
        lines.append("    classDef catbrowser fill:#fff3e0,stroke:#e65100")
        lines.append("    classDef catfile fill:#f3e5f5,stroke:#4a148c")
        lines.append("    classDef catapi fill:#ffebee,stroke:#b71c1c")
        
        return "\n".join(lines)
    
    def to_json(self) -> Dict:
        """导出为JSON（供D3.js等可视化库使用）"""
        nodes = [
            {
                "id": tool.id,
                "name": tool.name,
                "category": tool.category.value,
                "call_count": tool.call_count,
                "success_rate": tool.success_rate
            }
            for tool in self.graph.tools.values()
        ]
        
        edges = []
        for tool in self.graph.tools.values():
            for dep in tool.dependencies:
                edges.append({
                    "source": tool.id,
                    "target": dep,
                    "type": "depends_on"
                })
        
        return {"nodes": nodes, "edges": edges}

⚙️ 使用示例

from openclaw import CapabilityGraph, ToolNode, ToolCategory

# 1. 构建能力图谱
graph = CapabilityGraph()

# 2. 添加工具节点
graph.add_tool(ToolNode(
    id="web_search",
    name="Web Search",
    description="网络搜索工具",
    category=ToolCategory.SEARCH,
    inputs=["query"],
    outputs=["results"]
))

graph.add_tool(ToolNode(
    id="browser",
    name="Browser",
    description="浏览器自动化",
    category=ToolCategory.BROWSER,
    inputs=["url"],
    outputs=["page_content"],
    dependencies={"web_search"}  # 依赖web_search
))

graph.add_tool(ToolNode(
    id="content_extractor",
    name="Content Extractor",
    description="内容提取",
    category=ToolCategory.DATA,
    inputs=["html"],
    outputs=["text"],
    dependencies={"browser"}
))

graph.add_tool(ToolNode(
    id="summarizer",
    name="Summarizer",
    description="文本摘要",
    category=ToolCategory.DATA,
    inputs=["text"],
    outputs=["summary"],
    dependencies={"content_extractor"}
))

# 3. 添加依赖关系
graph.add_dependency("browser", "web_search")
graph.add_dependency("content_extractor", "browser")
graph.add_dependency("summarizer", "content_extractor")

# 4. 调用链追踪
tracker = CallChainTracker(graph)
tracker.start_chain("chain-1", "web_search")
tracker.add_call("chain-1", "browser")
tracker.add_call("chain-1", "content_extractor")
tracker.add_call("chain-1", "summarizer")
chain = tracker.end_chain("chain-1")

# 5. 分析调用链
analysis = tracker.analyze_chain(chain)
print(f"调用链分析: {analysis}")

# 6. 可视化
visualizer = GraphVisualizer(graph)
mermaid_code = visualizer.to_mermaid()
print(f"\nMermaid图:\n{mermaid_code}")

# 7. 导出JSON（用于前端可视化）
json_data = visualizer.to_json()
print(f"\nJSON数据: {len(json_data['nodes'])} 个节点, {len(json_data['edges'])} 条边")

🏆 最佳实践

🔗 相关链接

• 工具调用基础教程
• 工具链编排
• 工具编排实战
• 工具权限管理
• Skills依赖管理
• Mermaid.js 可视化文档