0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@hidemi-k(hidemi k)

Microsoft Agent Framework (MAF) rc5 入門 ― Groq × Llama-3 でネットワーク分析エージェントを作る

0
Posted at

はじめに

本連載では、MAF rc5 × eBPF/XDP を組み合わせた自律型 SASE フレームワーク（hidemi-k/maf-ebpf-sase）の設計と実装を、モジュールごとに解説していきます。まずは全体を支える基盤として Microsoft Agent Framework (MAF) rc5 にフォーカスし、「なぜ MAF を選んだか」「どう使うのか」を実装ベースで解説します。コードはリポジトリの netmiko_agent_framework.ipynb から抜粋しています。

ターゲット読者：

  • LangChain や AutoGen は触ったことがあるが、MAF は初めて
  • ネットワーク・インフラ制御にエージェントを使いたい
  • Groq の無料枠で手軽に試したい

※本記事の内容は個人の研究成果であり、所属組織の業務や公式見解とは一切関係ありません。

MAF とは何か

Microsoft Agent Framework (MAF) は、AutoGen と Semantic Kernel を統合する形で生まれた Microsoft のエージェントフレームワークです。

主要なオーケストレーション方式として以下をサポートしています：

モード 概要
Sequential ステップバイステップのワークフロー
Concurrent 複数エージェントの並列実行
Group Chat エージェント間の協調対話
Handoff タスクを次のエージェントへ引き継ぎ
Magnetic オーケストレーターがタスクリストを動的に管理

本プロジェクトでは、インフラ制御の特性上 Handoff / Magnetic パターン を中心に活用しています。

なぜ MAF rc5 を選んだか

LangChain・CrewAI・AutoGen など複数のフレームワークを検討した上で MAF rc5 を採用しました。インフラ制御という用途で決め手になったポイントを 4 つ挙げます。

① Native Tool Dispatch ― ディスパッチャを手書きしなくていい

MAF は関数シグネチャと Docstring からツールを自動生成します。LangChain では @tool デコレーターや Tool オブジェクトのラップが必要ですが、MAF では関数をそのまま tools=[] に渡すだけです。ツールが増えるほどこの差が効いてきます。

② Session & History Management ― 会話履歴の管理を委譲できる

複数ステップにわたる診断フローでは、過去のやり取りを保持しながら LLM に推論させる必要があります。MAF のセッション層がトークン上限の管理も含めて担ってくれるため、エージェントのロジックに集中できます。

③ OpenAI 互換インターフェース ― Groq の Llama-3 と繋がる

OpenAIChatClient に base_url を指定するだけで、Groq 上の Llama-3 を使えます。Azure OpenAI への切り替えもアプリ層を変えずに対応可能です。後述の make_client() がその実装例です。

④ Consistent Error Handling ― レート制限・リトライを一元管理

Groq の無料枠はレート制限が厳しいですが、MAF がリトライ処理をフレームワーク層で共通化しています。エージェントごとに個別実装する必要がありません。

インストール

pip install agent-framework netmiko openai

2026年3月現在、MAF はプレビュー段階（rc5）です。

Step 1: インポートと API キー読み込み

OpenAIChatClient は agent_framework のトップレベルではなく、agent_framework.openai サブモジュールからインポートする点に注意してください。

import asyncio
import os
import configparser
from typing import List, Dict, Any
from datetime import datetime

# Netmiko
from netmiko import ConnectHandler
from netmiko.exceptions import AuthenticationException, NetMikoTimeoutException, SSHException

# agent-framework rc5
# ★ OpenAIChatClient は agent_framework.openai サブモジュールからインポート
#   （agent_framework トップレベルには存在しない）
from agent_framework import Agent, Message
from agent_framework.openai import OpenAIChatClient  # ★ 正しいimportパス

print("✅ インポート完了")
print("   OpenAIChatClient: agent_framework.openai から正しくインポート")

API キーの読み込みは config.ini と環境変数の両方に対応しています：

config = configparser.ConfigParser()
config_paths = [
    '/home/admin/config/config.ini',
    './config.ini',
    os.path.expanduser('~/config/config.ini')
]

GROQ_API_KEY = None
for path in config_paths:
    if os.path.exists(path):
        config.read(path)
        if 'GROQ' in config and 'GROQ_API_KEY' in config['GROQ']:
            GROQ_API_KEY = config['GROQ']['GROQ_API_KEY']
            print(f"✅ APIキーを {path} から読み込みました")
            break

if not GROQ_API_KEY:
    GROQ_API_KEY = os.getenv('GROQ_API_KEY')
    if GROQ_API_KEY:
        print("✅ APIキーを環境変数から読み込みました")
    else:
        print("⚠️  APIキーが見つかりません - 手動で設定してください")

Step 2: OpenAIChatClient ファクトリ関数

MAF rc5 以前は BaseChatClient を継承した GroqChatClient を約 40 行で手動実装する必要がありました。rc5 からは OpenAIChatClient をそのまま使えます。Groq は OpenAI 互換 API のため base_url を差し替えるだけです。

# ★ v1: GroqChatClient(BaseChatClient) 手動実装 → 廃止
#        _inner_get_response・型変換を自前で書く必要があった

# ★ v2: OpenAIChatClient のファクトリ関数に置き換え
GROQ_BASE_URL = "https://api.groq.com/openai/v1"
DEFAULT_MODEL = "llama-3.3-70b-versatile"

def make_client(model_id: str = DEFAULT_MODEL) -> OpenAIChatClient:
    """
    MAFネイティブ OpenAIChatClient を生成する。

    Groq は OpenAI 互換 API のため base_url を差し替えるだけで流用可能。
    BaseChatClient の手動実装（_inner_get_response・型変換）は一切不要。

    引数名は MAF 公式ドキュメント準拠:
      model_id  : モデル名（'model' ではなく 'model_id'）
      api_key   : APIキー
      base_url  : Groq互換エンドポイント
    """
    return OpenAIChatClient(
        model_id=model_id,       # ★ 'model' ではなく 'model_id'
        api_key=GROQ_API_KEY,
        base_url=GROQ_BASE_URL,
    )

print("✅ make_client() 定義完了")
print(f"   エンドポイント  : {GROQ_BASE_URL}")
print(f"   デフォルトモデル: {DEFAULT_MODEL}")

model_id というキーワード引数は model ではありません。MAF の仕様に合わせる必要があります。

Step 3: エージェントの定義

エージェントは Agent クラスに name・client・instructions を渡すだけです。make_client() を各エージェントに渡すことで、全エージェントが同じ LLM バックエンドを使いながら独立したセッションを持ちます。

agent = Agent(
    name=config['name'],
    client=make_client(self.model),  # エージェントごとに独立したクライアント
    instructions=instructions
)

本実装では役割の異なる 4 種類のエージェント を定義しています：

エージェント 役割
設計担当 (Engineer1) 技術的な観点から分析
運用担当 (Engineer2) 運用上の観点から要点を整理
翻訳担当 (Translator) 英語出力を日本語に翻訳・補足
レビュー担当 (_Reviewer) Self-Correction 用。全分析を確認・指摘

有効にするエージェントは実行時に選択できます：

ENABLED_AGENTS = [
    'Engineer1',   # 設計担当
    'Engineer2',   # 運用担当
    # 'Translator',  # 翻訳担当（必要に応じて）
]

Step 4: Self-Correction ループ

v1 では分析が一方通行でしたが、v2 では Reviewer エージェントによるフィードバックループを追加しています。

[ラウンドロビン分析]
  🔧 設計担当 → 📋 運用担当 → ...
        ↓
[Self-Correction]
  👁️ レビュー担当が全分析を確認・誤りを指摘
        ↓ (指摘があった場合のみ)
  🔧 設計担当が指摘を踏まえて再分析

実装上の重要ポイントとして、暴走防止の max_calls があります。LLM エージェントはループに入ると API を呼び続けるリスクがあります。ネットワーク機器への誤操作に繋がりかねないため、コール数に上限を設けています：

def _check_call_limit(self) -> bool:
    """
    ★ 暴走防止チェック
    max_calls を超えたら False を返し、処理を中断する。
    ネットワーク機器への誤操作を引き起こす連鎖呼び出しを防ぐ。
    """
    self._call_count += 1
    if self._call_count > self.max_calls:
        print(f"⛔ [暴走防止] APIコール数が上限({self.max_calls})を超えました。処理を中断します。")
        return False
    return True

max_calls の目安は エージェント数 × MAX_TURNS × コマンド数 + Reviewer 分 です。

Step 5: Netmiko でコマンド実行

Netmiko でネットワーク機器に接続し、コマンド出力を取得します。認証エラー・タイムアウト・SSH エラーはそれぞれ個別にハンドリングしています：

def run_netmiko_commands(
    ip: str,
    port: str,
    username: str,
    password: str,
    device_type: str,
    commands: List[str]
) -> List[tuple]:
    """
    Netmikoでネットワーク機器に接続してコマンドを実行

    Returns:
        List[tuple]: [(command, output), ...]
    """
    try:
        device = {
            "device_type": device_type,
            "port": port,
            "ip": ip,
            "username": username,
            "password": password,
        }

        print(f"🔌 {ip} に接続中...")
        connection = ConnectHandler(**device)

        results = []
        for command in commands:
            print(f"  📝 実行中: {command}")
            output = connection.send_command(command, read_timeout=120)
            results.append((command, output))

        connection.disconnect()
        print(f"✅ {len(commands)} 個のコマンドを実行完了")
        return results

    except AuthenticationException:
        return [("認証エラー", "ユーザー名またはパスワードが正しくありません。")]
    except NetMikoTimeoutException:
        return [("接続タイムアウト", "指定されたIPアドレスまたはポートへの接続に失敗しました。")]
    except SSHException as e:
        return [("SSHエラー", f"SSH接続中にエラーが発生しました: {str(e)}")]
    except Exception as e:
        return [("予期せぬエラー", f"Netmikoの実行中にエラーが発生しました: {type(e).__name__}: {str(e)}")]

Step 6: 実行

デバイス設定とコマンドを定義して実行します：

DEVICE_CONFIG = {
    'ip':          '192.0.2.1',   # 接続先 IP
    'port':        '22',
    'username':    'admin',
    'password':    'YOUR_PASSWORD_HERE',
    'device_type': 'juniper_junos'
}

COMMANDS = [
    'show version',
    'show bgp summary',
    'show interfaces brief',
]

MAX_CALLS             = 20    # APIコール上限
ENABLE_SELF_CORRECTION = True

# ステップ1: Netmiko でコマンド実行
netmiko_results = run_netmiko_commands(
    ip=DEVICE_CONFIG['ip'],
    port=DEVICE_CONFIG['port'],
    username=DEVICE_CONFIG['username'],
    password=DEVICE_CONFIG['password'],
    device_type=DEVICE_CONFIG['device_type'],
    commands=COMMANDS
)

# ステップ2: MAF エージェントで分析
team = NetworkAnalysisTeam(
    enabled_agents=ENABLED_AGENTS,
    max_calls=MAX_CALLS,
    enable_self_correction=ENABLE_SELF_CORRECTION,
    model=DEFAULT_MODEL,
)

all_analyses = []
for cmd, output in netmiko_results:
    analysis = await team.analyze_command_output(
        command=cmd,
        output=output,
        user_task="ネットワークの状態を分析し、問題点があれば指摘してください。",
        max_turns=1
    )
    all_analyses.append({'command': cmd, 'analysis': analysis})

print(f"\n✅ 全ての分析が完了しました")
print(f"   総APIコール数: {team._call_count} / {team.max_calls}")

v1 → v2 の変化まとめ

改修 v1 v2
① クライアント GroqChatClient(BaseChatClient) 手動実装（約40行） OpenAIChatClient MAFネイティブ
② 状態管理 グローバル変数 + Native型変換 インスタンス変数 List[Message]
③ 暴走防止 なし max_calls でAPIコール上限
④ Self-Correction なし（一方通行） Reviewer → Generator フィードバックループ

最も大きな変化は ① の OpenAIChatClient 移行です。BaseChatClient の手動実装が不要になったことで、MAF が提供する会話履歴管理・エラーハンドリングをそのまま享受できるようになりました。

他フレームワークとの比較（個人的な所感）

観点 MAF rc5 LangChain CrewAI
Tool 定義の簡潔さ ◎ 関数をそのまま渡す △ デコレーター必要 △ 専用クラス必要
セッション管理 ◎ フレームワーク層 △ 自前実装 ○
マルチエージェント ◎ 多様なパターン ○ ◎
ドキュメント量 △ まだ少ない ◎ ○
本番実績 △ プレビュー段階 ◎ ○

ドキュメントがまだ少ない点は正直デメリットです。ただし、インフラ制御のように「会話履歴が重要で、エラーハンドリングを確実にしたい」用途では MAF の設計思想がよくマッチしています。

まとめ

MAF rc5 × Netmiko の組み合わせで、ネットワーク機器のコマンド出力を複数エージェントが協調して分析するシステムを実装しました。ポイントをまとめると：

  • OpenAIChatClient は agent_framework.openai からインポートする（トップレベルにはない）
  • model_id というキーワード引数に注意（model ではない）
  • max_calls で暴走防止を必ず入れる（インフラ操作では特に重要）
  • Reviewer エージェントによる Self-Correction で分析精度を上げる

次回は network_diagnostic_agent_v5.ipynb をベースに、5種の専門エージェントが L2/L3 を横断的に分析するマルチレイヤ障害診断システムの実装を紹介します。その先では、MAF による判断を eBPF/XDP によるラインレートのパケット遮断にどう直結させるか――「脳と筋肉の接続」についても解説していく予定です。

参考リンク

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?