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

はじめてのエージェント

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.29

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

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

config.tomlToml
[database]
host = "localhost"
port = 5432
database = "agenticstar"
username = "agent_user"
password = "your_password"
pool_min_size = 2
pool_max_size = 10
# Azure AD 認証を使う場合のみ true にし、[database.azure_ad] を設定
use_azure_ad = false

[events]
enable_db_handler = true
enable_webhook_handler = false

PostgreSQLConfig.from_toml() は既定で [database] セクションを読み込みます。フィールド名は username / pool_min_size / pool_max_size です。Azure AD 認証を使う場合は [database] 直下に use_azure_ad = true を置き、[database.azure_ad] に tenant_id / client_id / client_secret を設定します。

ステップ 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 のレスポンスがコンソールに出力されます。

Marketplace 互換で実行する（runner・SDK 0.5.29+）​

ローカルで動いた agent 関数は、run_marketplace_agent に渡すだけで Marketplace 互換の終端ライフサイクルで実行できます。identity（EXECUTION_ID 等）の受領・検証、 入力メッセージの取得、結果の DB 保存、Webhook 通知、終端イベント（何が起きても 正確に 1 回）、cleanup は runner が担うため、上のステップ 3 で書いたような main ボイラープレートは不要になります。

marketplace_agent.pyPython
from agenticstar_platform import run_marketplace_agent


async def my_agent(emitter, message: str) -> str:
    """エージェント本体。ロジックは自由（LangChain / OpenAI Agents SDK / 自作）。"""
    return message.upper()  # ここを実際の処理に置き換える


if __name__ == "__main__":
    run_marketplace_agent(my_agent)
curl
pip install 'agenticstar-platform[runner]==0.5.29'
python marketplace_agent.py

プラットフォーム上では EXECUTION_ID / CONVERSATION_ID / USER_ID / MESSAGE_ID は Marketplace executor が Pod 起動時に注入します。DB 接続 （DB_HOST 等、PostgreSQLConfig.from_env() 契約）と WEBHOOK_URL は エージェント登録時の環境変数として設定します。必須変数が欠けている場合、 runner は agent を呼ばずに MarketplaceRunnerConfigError で停止します。

  • agent 関数の戻り値が completion_success として保存・通知されます
  • agent 関数の例外は completion_failure に収束します（traceback はログのみ）
  • agent 関数が自分で終端イベントを emit する形でも二重送信にはなりません

ステップ 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 生成器と Qdrant マネージャを初期化
    # QdrantManager は EmbeddingGenerator を必須引数に取る（検索時に内部で埋め込みを生成）
    embed_config = EmbeddingConfig.from_toml("config.toml")
    generator = EmbeddingGenerator(embed_config)

    qdrant_config = QdrantConfig.from_toml("config.toml")  # collection_name は設定から
    qdrant = QdrantManager(qdrant_config, generator)
    await qdrant.initialize()

    # クエリテキストで類似検索（query_vector は不要。埋め込みは内部生成）
    result = await qdrant.search(query_text=query, limit=top_k)

    # 戻り値は dict。ヒットは result["data"]["results"] に格納される
    return [
        {"content": hit["payload"].get("content", ""), "score": hit["score"]}
        for hit in result["data"]["results"]
    ]

プロジェクト構成の例​

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

次のステップ​