Kagura AI: マルチエージェント構築フレームワーク

Posted at 2024-12-09

Kagura AI: YAMLベースの軽量AIエージェントフレームワーク

こんにちは！今回は、私が開発したオープンソースのAIエージェントフレームワーク「Kagura AI」について紹介させていただきます。

開発の背景

最近、「AI Agents」という概念に注目が集まっています。これは特定のタスクや目的に特化したAIアシスタントを指すものですが、エンジニアの視点からみると、以下のような課題があります：

実装の複雑さ：高い自由度がある反面、エージェントの実装が煩雑になりがち
状態管理の難しさ：複数のエージェントを連携させる際の状態管理が複雑
デバッグの困難さ：エラーハンドリングやデバッグが困難な場面が多い

これらの課題を解決するため、より手軽にAIエージェントを構築・管理できるフレームワークとして、Kagura AIを開発しました。

対象者

複数のAIエージェントを連携させたい
型安全性を保ちながら柔軟な開発をしたい
YAMLベースの設定で簡潔に実装したい
エラーハンドリングやデバッグを効率化したい

Kagura AIの特徴

1. YAMLベースの簡潔な設定

Kagura AIの最大の特徴は、YAMLファイルによる直感的な設定です。主要な設定フィールドは以下の通りです：

フィールド	説明	必須	例
`type`	エージェントの種類（atomic/tool/workflow）	✓	`atomic`
`description`	エージェントの説明（多言語対応）	✓	下記参照
`instructions`	LLMへの指示（多言語対応）	✓	下記参照
`prompt`	プロンプトテンプレート（多言語対応）	✓	下記参照
`response_fields`	応答フィールドの定義	-	`["summary", "keywords"]`
`input_fields`	入力フィールドの定義	-	`["content", "max_length"]`
`llm`	LLM設定（model, max_tokens等）	-	下記参照
`custom_tool`	カスタムツールのパス	-	`tools.process_data`

LLM設定オプション

オプション	説明	デフォルト値
`model`	使用するLLMモデル	`openai/gpt-4o-mini`
`max_tokens`	最大トークン数	4096
`retry_count`	リトライ回数	3
`stream`	ストリーミング有効化	`false`

設定例：

Kagura AIの最大の特徴は、YAMLファイルによる直感的な設定です。以下は基本的なエージェントの設定例です：

# agent.yml
type: atomic  # atomic, tool, workflow のいずれか
description:
  - language: en
    text: "This agent summarizes documents."
  - language: ja
    text: "文書を要約するエージェントです。"
instructions:
  - language: en
    description: |
      Summarize the given document clearly and concisely.
prompt:
  - language: en
    template: |
      Summarize this document:
      {content}
response_fields:
  - summary
input_fields:
  - content

2. 3種類のエージェントタイプ

Kagura AIには3つの基本的なエージェントタイプがあります：

Atomic Agent
- LLMと対話する基本的なエージェント
- 型安全な状態管理
- プロンプトテンプレート機能
Tool Agent
- LLMを使用せず、データ処理や外部API連携に特化
- 高速な実行
- カスタム処理の実装が容易
Workflow Agent
- 複数のエージェントを組み合わせたワークフロー
- 状態の受け渡しを自動管理
- 条件分岐やエラーハンドリング

3. 型安全な状態管理

状態管理はPydanticモデルを使用し、型安全性を確保：

# state_model.yml
custom_models:
  - name: Summary
    fields:
      - name: text
        type: str
        description:
          - language: en
            text: "Summarized text content"
          - language: ja
            text: "要約されたテキスト内容"
      - name: keywords
        type: List[str]
        description:
          - language: en
            text: "Key topics from the text"
state_fields:
  - name: content
    type: str
  - name: summary
    type: Summary

実践的な使用例

1. シンプルなチャットボットの作成

from kagura.core.agent import Agent

async def chat():
    # エージェントの初期化
    agent = Agent.assigner("chat")
    
    # メッセージの送信と応答のストリーミング
    async for response in await agent.execute("こんにちは！"):
        print(response, end="")

if __name__ == "__main__":
    import asyncio
    asyncio.run(chat())

2. ドキュメント処理ワークフローの構築

# agents/document_processor/agent.yml
type: workflow
entry_point: content_fetcher
nodes:
  - content_fetcher    # URLからコンテンツを取得
  - text_converter     # テキストに変換
  - summarizer         # 要約を生成

edges:
  - from: content_fetcher
    to: text_converter
  - from: text_converter
    to: summarizer

state_field_bindings:
  - from: text_converter.converted_content
    to: summarizer.content.text

async def process_document(url: str):
    agent = Agent.assigner("document_processor")
    async for state in await agent.execute({"url": url}):
        print(f"現在の処理: {state.AGENT.agent_name}")

3. カスタムツールエージェントの実装

# tools.py
from kagura.core.models import StateModel

async def process_data(state: StateModel) -> StateModel:
    """データ処理用のカスタムツール"""
    try:
        # カスタム処理をここに実装
        processed_data = await external_api_process(state.input_data)
        state.output_data = processed_data
        return state
    except Exception as e:
        raise Exception(f"処理エラー: {str(e)}")

# agent.yml
type: tool
custom_tool: tools.process_data
response_fields:
  - output_data
input_fields:
  - input_data

インストールと使い方

インストール

git clone https://github.com/JFK/kagura-ai.git
cd kagura-ai
uv add kagura-ai

# or

pip install kagura-ai

# kagura commandを実行して初期化する
kagura

基本的な使い方

エージェントディレクトリの作成：

mkdir -p ~/.config/kagura/agents/my_agent

設定ファイルの作成：

# ~/.config/kagura/agents/my_agent/agent.yml
type: atomic
description:
  - language: ja
    text: "カスタムエージェントです。"
instructions:
  - language: ja
    description: |
      あなたは特定のタスクを実行するエージェントです。
prompt:
  - language: ja
    template: |
      {query}

エージェントの使用：

from kagura.core.agent import Agent

async def run_agent():
    agent = Agent.assigner("my_agent")
    result = await agent.execute("こんにちは")
    print(result)

asyncio.run(run_agent())

エラーハンドリングとデバッグ

Kagura AIは包括的なエラーハンドリングとデバッグ機能を提供します：

try:
    result = await agent.execute(input_data)
    if not result.SUCCESS:
        print(f"エラー: {result.ERROR_MESSAGE}")
except Exception as e:
    print(f"実行エラー: {str(e)}")

また、詳細なログを有効にすることも可能です：

export LOG_LEVEL=DEBUG

開発への参加

Kagura AIはオープンソースプロジェクトとして、コミュニティの皆様の協力を歓迎しています。以下のような形で開発に参加いただけます：

コントリビューション方法

機能の提案や改善
- GitHubのIssuesで新機能の提案
- 既存機能の改善案の共有
- ドキュメントの改善提案
コードの貢献
- バグ修正のPull Request
- 新機能の実装
- テストの追加や改善
エージェントの共有
- 独自に開発したエージェントの公開
- 実用的なユースケースの共有
- サンプルコードの提供

参加するには

CONTRIBUTING_JA.mdを確認
GitHubリポジトリをフォーク

開発環境のセットアップ:

git clone https://github.com/your-username/kagura-ai.git
cd kagura-ai
uv sync --frozen --all-extras --dev

より詳しい情報は開発者ガイドをご覧ください。

参考リンク

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up