メインコンテンツまでスキップ

はじめてのエージェント

agenticstar-platform SDK を使って、最小構成のカスタムエージェントを作成・実行するまでの手順です。SDK はインフラ（DB・イベント・ストレージ等）を提供し、エージェントロジックは開発者が自由に設計します。

前提条件​

  • Python 3.12 以上
  • pip または uv
  • PostgreSQL データベース（ローカルまたは Azure Database for PostgreSQL）
  • Azure OpenAI API キー（LLM 呼び出し用）

ステップ 1: SDK のインストール​

curl
pip install agenticstar-platform[all]==0.5.3

ステップ 2: 設定ファイルの作成​

config.toml を作成します。SDK の各モジュールはこの設定ファイルから構成を読み取ります。

config.tomlToml
[postgresql]
host = "localhost"
port = 5432
database = "agenticstar"
user = "agent_user"
password = "your_password"
min_connections = 2
max_connections = 10

[postgresql.azure_ad]
enabled = false

[events]
enable_db_handler = true
enable_webhook_handler = false

ステップ 3: エージェントの実装​

my_agent.py を作成します。この例では、SDK の DB モジュール でジョブ情報を取得し、Events モジュール で進捗をストリーミング配信するシンプルなエージェントを実装します。

my_agent.pyPython
import asyncio
from openai import AsyncAzureOpenAI

from agenticstar_platform.db import PostgreSQLManager, PostgreSQLConfig, DataAccess
from agenticstar_platform.db.execution_access import ExecutionAccess
from agenticstar_platform.events import EventEmitter, EventType

# --- 設定 ---
db_config = PostgreSQLConfig.from_toml("config.toml")
llm_client = AsyncAzureOpenAI(
    azure_endpoint="https://your-endpoint.openai.azure.com/",
    api_version="2024-12-01-preview",
    api_key="your-api-key",
)

async def run_agent(execution_id: str, user_message: str):
    """最小構成のエージェント実行フロー"""

    # 1. DB 接続を初期化
    db_manager = PostgreSQLManager(db_config)
    da = DataAccess(db_manager)
    await da.initialize()

    execution_access = ExecutionAccess(da)

    # 2. イベントエミッター初期化
    emitter = EventEmitter(execution_id=execution_id)

    # 3. 実行開始イベント
    await emitter.emit_event(
        event_type=EventType.PHASE_START,
        message="リクエストを処理中...",
        metadata={"phase": "processing"},
    )

    # 4. 過去のメッセージを取得
    history = await execution_access.get_messages(execution_id=execution_id)

    # 5. LLM 呼び出し（エージェントロジックは自由に設計）
    messages = [{"role": "system", "content": "あなたは親切なアシスタントです。"}]
    if history:
        for h in history:
            messages.append({"role": h["role"], "content": h["content"]})
    messages.append({"role": "user", "content": user_message})

    response = await llm_client.chat.completions.create(
        model="gpt-4.1",
        messages=messages,
    )
    answer = response.choices[0].message.content

    # 6. 完了イベント
    await emitter.emit_event(
        event_type=EventType.COMPLETION_SUCCESS,
        message=answer,
    )

    # 7. クリーンアップ
    await emitter.cleanup()
    await da.close()

    return answer

if __name__ == "__main__":
    result = asyncio.run(run_agent("exec-001", "こんにちは！"))
    print(result)

ステップ 4: 動作確認​

curl
python my_agent.py

正常に動作すると、LLM のレスポンスがコンソールに出力されます。

ステップ 5: ツールの追加（オプション）​

エージェントにツール（関数呼び出し）を追加する例です。SDK の RAG モジュール を使ってナレッジベースを検索します。

tools/search_knowledge.pyPython
from agenticstar_platform.rag import QdrantManager, QdrantConfig
from agenticstar_platform.rag import EmbeddingGenerator, EmbeddingConfig

async def search_knowledge(query: str, top_k: int = 5) -> list[dict]:
    """ナレッジベースからクエリに関連するドキュメントを検索"""

    # Embedding 生成
    embed_config = EmbeddingConfig.from_toml("config.toml")
    generator = EmbeddingGenerator(embed_config)
    query_vector = await generator.generate(query)

    # Qdrant で類似検索
    qdrant_config = QdrantConfig.from_toml("config.toml")
    qdrant = QdrantManager(qdrant_config)

    results = await qdrant.search(
        collection_name="knowledge_base",
        query_vector=query_vector,
        limit=top_k,
    )

    return [
        {"content": r.payload.get("content", ""), "score": r.score}
        for r in results
    ]

プロジェクト構成の例​

my-agent/
├── config.toml              # SDK 設定ファイル
├── my_agent.py              # エージェントのメインロジック
├── tools/
│   ├── search_knowledge.py  # RAG 検索ツール
│   └── file_manager.py      # ストレージ操作ツール
├── Dockerfile               # コンテナビルド用
├── requirements.txt
└── README.md

次のステップ​