Draw.io で描いた図をそのままAIプロンプトに — ビジュアルエージェント設計入門

<mxGraphModel>
  <root>
    <mxCell id="0"/>
    <mxCell id="1" parent="0"/>
    <!-- 図形要素 -->
    <mxCell id="2" value="Google検索を実行"
            style="rounded=1;whiteSpace=wrap;"
            vertex="1" parent="1">
      <mxGeometry x="100" y="100" width="200" height="60" as="geometry"/>
      <!-- カスタムメタデータ -->
      <object label="Google検索を実行"
              SystemPrompt="指定されたキーワードでGoogle検索を行い、上位3件の結果を返す"
              ToolName="web_search"/>
    </mxCell>
  </root>
</mxGraphModel>

import xml.etree.ElementTree as ET
 
def parse_drawio_prompt(filepath):
    """Draw.ioファイルからプロンプト情報を抽出"""
    tree = ET.parse(filepath)
    root = tree.getroot()
 
    tasks = []
    for cell in root.iter("mxCell"):
        value = cell.get("value", "")
        # object要素を持つセルを探す
        obj = cell.find("object")
        if obj is not None:
            task = {
                "label": obj.get("label", value),
                "system_prompt": obj.get("SystemPrompt", ""),
                "tool": obj.get("ToolName", ""),
                "id": cell.get("id")
            }
            tasks.append(task)
 
    return tasks
 
# 使用例
tasks = parse_drawio_prompt("prompt.dio")
for task in tasks:
    print(f"Task: {task['label']}")
    print(f"  Prompt: {task['system_prompt']}")
    print(f"  Tool: {task['tool']}")
# 出力例:
# Task: Google検索を実行
#   Prompt: 指定されたキーワードでGoogle検索を行い、上位3件の結果を返す
#   Tool: web_search

def parse_drawio_workflow(filepath):
    """Draw.ioファイルからワークフロー（ノード+エッジ）を抽出"""
    tree = ET.parse(filepath)
    root = tree.getroot()
 
    nodes = {}
    edges = []
 
    for cell in root.iter("mxCell"):
        cell_id = cell.get("id", "")
        source = cell.get("source")
        target = cell.get("target")
 
        if source and target:
            # エッジ（矢印）
            edges.append({"from": source, "to": target})
        elif cell.get("vertex") == "1":
            # ノード（図形）
            obj = cell.find("object")
            if obj is not None:
                nodes[cell_id] = {
                    "label": obj.get("label", cell.get("value", "")),
                    "prompt": obj.get("SystemPrompt", ""),
                    "tool": obj.get("ToolName", "")
                }
 
    return nodes, edges
 
def build_execution_order(nodes, edges):
    """エッジ情報から実行順序を決定"""
    # 入次数0のノードを開始点とするトポロジカルソート
    in_degree = {nid: 0 for nid in nodes}
    adjacency = {nid: [] for nid in nodes}
 
    for edge in edges:
        if edge["to"] in in_degree:
            in_degree[edge["to"]] += 1
        if edge["from"] in adjacency:
            adjacency[edge["from"]].append(edge["to"])
 
    queue = [nid for nid, deg in in_degree.items() if deg == 0]
    order = []
 
    while queue:
        current = queue.pop(0)
        order.append(current)
        for neighbor in adjacency.get(current, []):
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
 
    return order
 
# 使用例
nodes, edges = parse_drawio_workflow("sequence_prompt.dio")
order = build_execution_order(nodes, edges)
for node_id in order:
    node = nodes[node_id]
    print(f"Execute: {node['label']} (tool: {node['tool']})")
# 出力例:
# Execute: RSSフィード取得 (tool: rss_fetch)
# Execute: 条件フィルタリング (tool: filter)
# Execute: データ抽出 (tool: extract)
# Execute: 100字要約 (tool: summarize)

import asyncio
 
async def execute_agent(agent_name, tasks, nodes):
    """単一エージェントのタスクを順次実行"""
    results = []
    for task_id in tasks:
        node = nodes[task_id]
        print(f"[{agent_name}] Executing: {node['label']}")
        # 実際にはここでAI APIを呼び出す
        result = f"Result of {node['label']}"
        results.append(result)
        await asyncio.sleep(0.1)  # API呼び出しの模擬
    return results
 
async def run_multi_agent_workflow(agents_config, nodes):
    """複数エージェントを並列実行し、結果を集約"""
    tasks = [
        execute_agent(name, task_ids, nodes)
        for name, task_ids in agents_config.items()
    ]
    all_results = await asyncio.gather(*tasks)
    return dict(zip(agents_config.keys(), all_results))
 
# 使用例
agents = {
    "Agent A（市場調査）": ["node_1", "node_2"],
    "Agent B（技術調査）": ["node_3", "node_4"]
}
# results = asyncio.run(run_multi_agent_workflow(agents, nodes))
print("Multi-agent workflow configured")
# 出力: Multi-agent workflow configured