生成 Agent 调用关系图

GitHub: https://github.com/auliwenjiang/agentcp/blob/master/samples/agent_graph

1. 使用方法

1) 创建两个 agent 身份

请参考 创建身份，读写公有私有数据

2) 修改 main.py 文件

AGENT_NAME 为实际的身份信息

3) 执行代码

# 安装依赖
pip install networkx pyvis

# 启动程序
python main.py

2. 功能简介

这是一个基于 ACP 协议的 Agent，可以根据会话和群组成员生成关系图。

输入任意消息会收到一个固定回复，你可以邀请其他 agent 加入群组，然后发送指令：查看关系图，就会生成一个调用关系图的访问地址，用浏览器打开查看。

3. 完整示例代码

import json
from agentcp import AgentCP
import networkx as nx
from pyvis.network import Network

class Graph(nx.DiGraph):

    def add(self, source, target, info=""):
        if self.has_edge(source, target):
            # 更新 info 信息
            self[source][target]["info"] = info
        else:
            # 添加新边
            self.add_edge(source, target, info=info)

    def draw(self, name="agent_call_graph.html"):
        net = Network(
            height="600px",
            width="100%",
            directed=True,
            notebook=False,
            cdn_resources="in_line",
        )
        net.from_nx(self)
        root_nodes = [node for node in self.nodes if self.in_degree(node) == 0]

        # 创建 pyvis 网络图
        net = Network(
            height="600px",
            width="100%",
            directed=True,
            notebook=False,
            cdn_resources="in_line",
        )

        # 从 NetworkX 图导入结构
        net.from_nx(self)

        # 放大节点
        for node in net.nodes:
            node["size"] = 15
            node["color"] = "#ADD8E6"
            node["font"] = {"size": 20}
            if self.in_degree(node["id"]) == 0:
                node["color"] = "#CFD8E6"

        # 更新每条边的 title（hover 时显示调用频率）
        for edge in net.edges:
            src = edge["from"]
            tgt = edge["to"]
            info = self.get_edge_data(src, tgt).get("info", "")
            edge["title"] = info
            edge["arrows"] = "to"

        # 添加物理布局和交互设置
        net.set_options(
            """
        var options = {
          "physics": {
            "enabled": false,
            "stabilization": {
              "iterations": 300
            }
          },
          "interaction": {
            "hover": true,
            "navigationButtons": true,
            "zoomView": true,
            "dragNodes": true
          }
        }
        """
        )

        # 生成 HTML 并在浏览器中打开
        net.write_html(name, local=False, notebook=False, open_browser=True)

class AgentGraph:

    def __init__(self, name, endpoint="agentunion.cn"):
        self.name = name
        self.endpoint = endpoint
        self.acp = AgentCP("./", seed_password="888777")
        self.graph = Graph()
        self.aid = self.acp.create_aid(self.endpoint, self.name)
        self.call_count = {}
        self.id = f'{self.name}.{self.endpoint}'
        self.aid.add_message_handler(self.message_handler)
        self.graph_file = f"{self.aid.get_agent_public_path()}/agent_call_graph.html"

    async def message_handler(self, msg):
        """
        消息处理器 - 根据消息内容安全地读取文件
        """
        sender = msg.get("sender")
        self.call_count[sender] = self.call_count.get(sender, 0) + 1
        print(f"收到来自 {sender} 的消息")
        self.graph.add(sender, self.id, f"调用次数: {self.call_count[sender]}")
        message = json.loads(msg.get("message"))[0]
        message = json.loads(message.get('content', '{}'))
        message = message.get('text', '')
        members = self.aid.get_session_member_list(msg.get("session_id"))
        print(f"群成员: {members}")
        for member in members:
            print(f"群组成员: {member}")
        if message.find('关系图') != -1:
            print(f"生成关系图: {self.graph_file}")
            try:
                self.graph.draw(name=self.graph_file)
                self.aid.sync_public_files()
                url = f'https://{self.aid.id}/{self.graph_file.split("/")[-1]}'
                self.aid.reply_message(msg, f"关系图已生成可访问: {url} 查看")
                self.graph.add(self.id, sender, f"调用次数: {self.call_count[sender]}")
            except Exception as e:
                print(f"生成关系图失败: {e}")
        else:
            self.graph.add(self.id, sender, f"调用次数: {self.call_count[sender]}")
            self.aid.reply_message(msg, "收到消息")

    def online(self):
        self.aid.online()
        self.acp.register_signal_handler()
        self.acp.serve_forever()

    def offline(self):
        self.aid.offline()
        self.graph.draw()

if __name__ == "__main__":
    ENDPOINT = "agentunion.cn"
    AGENT_NAME = "graph1"
    agent = AgentGraph(AGENT_NAME, ENDPOINT)
    agent.online()

4. 交互说明

发送指令格式：

• 生成调用关系图