LoginSignup
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?

@garyo

AIエージェント(Codex)でAIエージェント(Agent SDK)を作る

Posted at

AIエージェントのOpenAIのCodex CLIを使ってOpenAIのAgent SDKを使ったAIエージェントを作ってみます。

Q.「openAIのAgetSDKを使ってAIエージェントプログラムを作ってください。与えられた目的からタスクを考えてそれを解決するために実
行するAIエージェントプログラムを作るにはどのような要求仕様が必要ですか？」

A.

codex
AIエージェント（OpenAI Agent SDK）を使って「与えられた目的からタスクを自動生成し、実行して解決する」プログラムを作る
ためには、下記のような要求仕様（Requirements Specification）があると考えられます。

  1. 概要

  • 本システムは、ユーザーから与えられた「高レベルの目的（ゴール）」を受け取り、内部で必要なサブタスクに分解し
    、順次実行してゴール達成を支援するAIエージェントを提供する。

  • エージェントのコアには OpenAI の Agent SDK
    を利用し、プランナー→実行→検証→メモリ更新…というループで動作する。

  1. 利害関係者

  • エンドユーザー（ゴールを与えて結果を得たい人）

  • システム管理者／DevOps（デプロイ・運用）

  • 開発者（エージェントのロジックやプラグインを追加・拡張する）

  1. 前提条件／制約

  • OpenAI API キーが発行済みで、環境変数（例：OPENAI_API_KEY）に設定されている

  • Node.js（または Python）ランタイムの LTS バージョン

  • Agent SDK のバージョン（例：@openai/agent-sdk ^1.0.0）

  • ネットワーク経由でツール（検索エンジン、データベース、外部APIなど）呼び出しが可能

  1. 機能要件

  2. ゴール入力

  • ユーザーが自然言語（日本語／英語）で目的を入力できる
  1. タスク分解（プランニング）

  • 目的を達成するためのサブタスクの列挙（例：To‑Doリスト）

  • 各タスクの優先度や依存関係を判断

  1. タスク実行（アクション）

  • 必要に応じて外部ツール（HTTP API／DBクエリ／検索など）を呼び出す

  • 実行結果を受け取り、次タスクの判断に反映

  1. メモリ・コンテキスト管理

  • 途中経過や中間結果を保持（短期メモリ／長期メモリ）

  • 再実行時のリトライやフォールバックをサポート

  1. 結果出力

  • 最終的な処理結果やログをユーザーに返却

  • 実行済みタスクのレポート生成

  1. 非機能要件

  • レイテンシ：１タスクあたりの平均応答時間は n 秒以内

  • 可用性：99.9％ 以上

  • スケーラビリティ：同時並列エージェント数 M まで対応

  • セキュリティ：
    • API キーや認証情報は Vault／環境変数で厳格管理
    • ログに秘匿情報が残らない

  • 障害耐性：
    • 外部API障害時のリトライ／タイムアウト設定
    • 異常終了時の自動ロールバック

  1. 外部インターフェース

  • REST / WebSocket / CLI など、ユーザーからゴールを受け取る入口

  • Agent SDK → OpenAI ChatCompletion API

  • Agent → プラグイン／ツール呼び出しポイント

  • ロギング：JSON形式で ELK / Datadog へ送信

  1. システム構成（高レベル）
    ┌─────────┐ ┌───────────────┐
    │ User UI │◀────▶│ Agent Core │
    └─────────┘ │(Planner,Memory)│
    └───────────────┬┘
    │
    ┌──────────────────┴────────────┐
    │ OpenAI Agent SDK / API │
    └──────────────────┬────────────┘
    │
    ┌──────────┴──────────┐
    │ Tools & Plugins API │
    └─────────────────────┘
  2. 依存関係・ライブラリ

  • @openai/agent-sdk

  • axios（HTTP クライアント）

  • dotenv（環境変数ロード）

  • winston などのロギングライブラリ

  1. テスト要件

  • ユニットテスト：プランナー、メモリ管理、ツール呼び出しラッパー

  • 統合テスト：エンドツーエンドでのゴール→結果パス検証

  • モック：OpenAI API モック、外部ツールモック

  1. 運用・保守
  • モニタリング：エージェント処理件数、成功率、レイテンシ
  • アラート：エラー率閾値超過時に通知
  • バージョン管理：AgentSDK や Node.js のバージョンアップ対応
  • ドキュメント：利用手順、プラグイン追加手順、トラブルシューティング

――――――――――
付録：要求仕様テンプレート（YAMLイメージ）

project:
name: goal-driven-ai-agent
description: "ユーザーの目的から自動でタスク生成・実行するAIエージェント"

stakeholders:

  • user: "サービス利用者"
  • devops: "運用チーム"
  • dev: "開発チーム"

environments:
runtime: "Node.js >=16.x"
agentSdk: "@openai/agent-sdk ^1.0.0"
openaiApiKey: "env:OPENAI_API_KEY"

functional_requirements:
FR1: "自然言語入力でゴールを受け取る"
FR2: "ゴールを分解してタスクリスト生成"
FR3: "ツール呼び出しを実行して成果を取得"
FR4: "メモリに中間結果を保存／参照"
FR5: "最終レポートを出力"

nonfunctional_requirements:
NFR1: "平均応答時間 < 3秒"
NFR2: "可用性 99.9%"
NFR3: "ログは 90日間保持"

interfaces:

  • type: REST
    endpoint: POST /api/agent/run
  • type: WebSocket
    endpoint: /ws/agent
  • type: Plugin
    spec: https://...

dependencies:

  • name: "axios"
  • name: "dotenv"
  • name: "winston"

testing:
unit: true
integration: true

operations:
monitoring: ["latency", "error_rate"]
alerts: ["error_rate > 5%"]

———

上記仕様をベースに、Agent SDK の初期化→ツール登録→Planner／Executor
のループ実装、インターフェース層の構築、環境・運用設定を進めていく流れがスムーズに設計・開発できます。

Q.「最初の要求仕様をequipment.mdに書き出してください。また私はopenAIのAgent SDKを使いたいのでpythonで書いてください。」
A.

Requirements Specification for Goal-Driven AI AgentOverview
本システムは、ユーザーから与えられた「高レベルの目的（ゴール）」を受け取り、内部で必要なサブタスクに分解し、順次実行してゴールを達成するAIエージェントを提供する。
Agent SDK（Python）を利用し、プランニング→実行→検証→メモリ更新のループで動作する。
Stakeholders
エンドユーザー: ゴールを与えて結果を得る。
システム管理者／DevOps: デプロイ・運用。
開発者: エージェントのロジックやプラグインを追加・拡張。
Preconditions / Constraints
OpenAI API キーが発行済みで、環境変数 OPENAI_API_KEY に設定されている。
Python 3.8 以上。
openai パッケージ（関数呼び出し対応版）がインストール済み。
ネットワーク経由で外部 API 呼び出しが可能。
Functional Requirements
FR1: ユーザーが自然言語（日本語／英語）でゴールを入力できる。
FR2: 目的を達成するためのサブタスクをプランニングし、外部ツール（関数呼び出し）で実行。
FR3: サブタスク実行結果を取得し、次のプランニングに反映。
FR4: 中間結果をメモリ（短期／長期）に保存し、リトライやフォールバックをサポート。
FR5: 最終的な処理結果と実行ログをユーザーに出力。
Non-Functional Requirements
レイテンシ: サブタスクあたりの平均応答時間は 3秒以内。
可用性: 99.9%以上。
スケーラビリティ: 同時並列エージェント数 10 まで対応。
セキュリティ: API キーは環境変数で厳格管理。ログに秘匿情報を出力しない。
障害耐性: 外部 API 障害時はリトライ・タイムアウト機能を実装。
External Interfaces
CLI: python agent.py "" でゴールを受け取り、結果を標準出力。
OpenAI ChatCompletion API (function calling)。
Optional: Web/UI や REST API を将来追加可能。
System Architecture (概要)
User CLI ←→ Agent Core ←→ OpenAI API ←→ External Tools
Agent Core:
Planner: LLM でタスク分解。
Executor: 関数呼び出しでサブタスク実行。
Memory: 中間結果・履歴管理。
Dependencies / Libraries
openai (関数呼び出し対応版)
python-dotenv
Testing Requirements
ユニットテスト: プランニング、関数呼び出しラッパー。
統合テスト: E2E でゴール→実行→結果を検証。
モック: OpenAI API、外部ツールをモック。
Operations & Maintenance
モニタリング: レイテンシ、成功率、エラー率。
アラート: エラー率が閾値を超えた場合に通知。
ドキュメント: セットアップ手順、拡張手順、トラブルシューティング

出来たコード

import os
import sys
import asyncio
from dotenv import load_dotenv
from agents import Agent, Runner
import json

async def main():
    load_dotenv()
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        print("Please set OPENAI_API_KEY in .env")
        sys.exit(1)
    if len(sys.argv) < 2:
        print("Usage: python sdk_agent.py <goal>")
        sys.exit(1)
    goal = " ".join(sys.argv[1:])
    agent = Agent(
        name="assistant",
        model="gpt-4o-mini"
    )
    runner = Runner()
    # FR2: Planning stage - decompose goal into tasks
    from openai import OpenAI
    client = OpenAI(api_key=api_key)
    loop = asyncio.get_running_loop()
    def plan_call():
        return client.chat.completions.create(
            model=agent.model,
            temperature=0,
            messages=[
                {"role": "system", "content": (
                    "You are a planner that decomposes a user goal into a JSON array of tasks. "
                    "Allowed actions: write_file (to create files), run_command (to execute shell commands), and search (to lookup web information). "
                    "If the goal requests execution of a command (such as ping), use run_command. "
                    "If the goal requires information, use search. "
                    "For code generation, use write_file, then run_command to compile and execute. "
                    "Respond with only valid JSON array of task objects, without any additional text."
                )},
                {"role": "user", "content": goal}
            ]
        )
    plan_resp = await loop.run_in_executor(None, plan_call)
    try:
        tasks = json.loads(plan_resp.choices[0].message.content)
    except Exception:
        # Fallback to single-step
        tasks = [{"action": "run_command", "args": {"command": goal}}]
    # If planner suggests a run_command for non-shell goal, switch to search
    if len(tasks) == 1 and tasks[0].get('action') == 'run_command':
        cmd = tasks[0].get('args', {}).get('command', '')
        # heuristics: if command contains non-latin (Japanese) or spaces >5, use search
        if any(ord(c) > 127 for c in cmd) or len(cmd.split()) > 3:
            tasks = [{"action": "search", "args": {"query": goal}}]
    print(f"Planner: {len(tasks)} tasks generated.")
    # Execute each task sequentially
    import re
    for i, task in enumerate(tasks, 1):
        action = task.get("action")
        args = task.get("args", {})
        # Fallback: handle ping command if needed
        if action == "run_command":
            cmd = args.get("command", "")
            if not cmd and "ping" in goal.lower():
                # extract hostname after 'ping'
                m = re.search(r'ping\s+([\w\.-]+)', goal)
                if not m:
                    m = re.search(r'([\w\.-]+)\s*に.*?ping', goal)
                host = m.group(1) if m else goal
                args['command'] = f"ping -c 4 {host}"
                print(f"Ping fallback: command set to '{args['command']}'")
        print(f"[{i}/{len(tasks)}] {action} {args}")
        # Describe subtask as instruction for runner
        inst = f"Call {action} with args {args}"
        result = await runner.run(agent, inst)
        print(result.final_output)
    return

if __name__ == "__main__":
    asyncio.run(main())

Q.「良くなりました。sdk_agent.pyはequipment.mdの要求仕様を何パーセントくらい満たしていますか？」
A.「 codex
ざっくり見積もると、sdk_agent.py は equipment.md に対して次のようなカバー率になります。

• FR1（CLI から自然言語ゴールを入力）……✓（100％）
• FR2（プランニング→実行ループ）……部分的（関数呼び出しでコード生成→実行はやっているが、LLM
によるタスク分解ループは未実装）⇒約 30％
• FR3（サブタスク結果のフィードバック反映）……最低限（function‐calling の単一ループ）⇒約 20％
• FR4（メモリ管理）……未実装⇒ 0％
• FR5（最終結果＋実行ログ出力）……部分的（最終結果は出力するが詳細ログは出していない）⇒約 40％

非機能要件・テスト・運用要件は未実装のため 0％。

総合すると、機能要件カバー率でおよそ 20～30％ 程度、全体要求としては約 20％ 程度の実装状況と言えます。

次のステップとしては、

* FR2 を課題分解（プランナー）＋多段ループで強化
* FR3 のリトライ・条件分岐を導入
* FR4 のメモリ（チェイン・オブ・ソート）実装
* ロギング／テストを追加

といった部分を進めると、要求仕様をさらに満たせます。

Q.「FR2 を課題分解（プランナー）＋多段ループで強化をお願いします。」

といった感じで進めています。

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?