Agent Builderで作成したエージェントをchatkit-pythonで動かす方法

Agent Builder は TypeScript / Python のコードをエクスポートできますが、そのままでは実行できません。エージェントはサーバー側で動作し、ChatKit のサーバー実装は chatkit-python です。本記事では、Agent Builder の Python コードを chatkit-python で動かす方法を解説します。

Agent BuilderからPythonコードをエクスポートする

Agent Builderの右上の </>Code アイコンをクリックするとダイアログが開きます。タブ切り替えで Agents SDK を選択し、サブタブで Python を選択することでエージェントのPythonコードが表示されます。

コピーボタンでコピーしたコードを拡張子 .py で保存しておきます。

chatkit-pythonの実行環境をセットアップする

サンプルプロジェクトを動かしてみると理解が早いと思います。本記事ではopenai-chatkit-advanced-samplesに手を入れていこうと思います。

サンプルプロジェクト「openai-chatkit-advanced-samples」を起動する

動かすにはnpmとuvが必要なのでインストールしておきます。環境変数のOPENAI_API_KEYを設定します。

GitHubからopenai-chatkit-advanced-samplesのコードを取得します。

openai-chatkit-starter-appと違ってバックエンドとフロントエンドの2つを起動する必要があります。

以下のコマンドを実行し、

npm run backend
npm run frontend

http://127.0.0.1:5170にアクセスします。以下のように表示されればOKです。

openai-chatkit-advanced-samples には 他にも3つ のサンプルが含まれています。examples/ 配下の各プロジェクトでも、先ほどのコマンドで起動できます。

ChatKit連携の実装ポイント

respond関数でAgentを実行する（実装フロー）

respond は AsyncIterator[ThreadStreamEvent] を返す必要があります。サンプルでは Runner.run_streamed() と stream_agent_response() を用いて ストリーム出力を行い、イベントを yield しています。

サンプルコードの解説

サンプルではストリーム対応になっており、以下のようなコードです。

result = Runner.run_streamed(
    self.assistant,
    agent_input,
    context=agent_context,
)

async for event in stream_agent_response(agent_context, result):
    yield event
return

ChatKitにはrespond関数の実装を楽にするヘルパー関数が準備されています。stream_agent_responseを使うことでRunner.run_streamedの結果を簡単にChatKitに返すことができます。

Agent Builderで作成したコード

Agent Builderで作成したエージェントは以下のようなコードになっています。

class WorkflowInput(BaseModel):
  input_as_text: str

# Main code entrypoint
async def run_workflow(workflow_input: WorkflowInput):
  ～中略～
  create_task_result = {
    "output_text": create_task_result_temp.final_output.json(),
    "output_parsed": create_task_result_temp.final_output.model_dump()
  }
  return create_task_result

ポイントは2つ。

• ストリーム対応になっていない
• Agent Builderで出力がウィジェットを指定していてもJSONでの出力になっている

つまり、このコードをサンプルを参考に実行してもChatKitに結果を返すことはできませんし、ウィジェットも表示されません。

Agent Builderで作成したエージェントの実行

以下のようなコードでエージェントを実行できます。

result = await run_workflow(wf_input)

# ユーザーメッセージ 
result_item=UserMessageItem(
    id = _gen_id("msg"),
    thread_id = thread.id,
    created_at = datetime.now(),  
    type="user_message",
    content=[
        UserMessageTextContent(
            type="input_text",
            text="Action decision=true."
        )
    ],
    inference_options=InferenceOptions(
        tool_choice=None,
        model=None,
    )
)

# スレッドにアイテムを追加
yield ThreadItemAddedEvent(
    type="thread.item.added",
    item=result_item
)

# スレッドの完了
yield ThreadItemDoneEvent(item=result_item)

run_workflow() でエージェントを実行した結果はJSONです。JSONをThreadStreamEventに変換する必要があります。Streamであれば stream_agent_response() で簡単に変換できますが、結構面倒です。このコードではユーザーメッセージで返しており、ウィジェットでは返せていません。

ウィジェットとして結果を返す

ウィジェットで返すにはWidget Builderで作成したウィジェットのWidgetItemを作成する必要があります。OpenAIのドキュメントでは、WidgetRoot.model_validate_json(WIDGET_JSON_STRING)でウィジェットを作成できると書いてあります。

WIDGET_JSON_STRINGに何を設定すればよいか。それはWidget Builderで作成したウィジェットのコードにあります。ダウンロードした .widgetファイルを開くと以下のようになっています。このtemplate部分がWIDGET_JSON_STRINGです。

{
  "version": "1.0",
  "name": "flight",
  "template": "{\"type\":\"Card\",\"size\":\"sm\",～中略～{\"type\":\"Caption\",\"value\":{{ (arrivalTime) | tojson }},\"children\":[]}]}]}]}]}]}",
  "jsonSchema": {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "widget",
    "type": "object",
～中略～  
}

しかし、このtemplate部分はそのまま使えません。上記のコードだとCaptionのvalue値が {{ (arrivalTime) | tojson }} とプレースフォルダーになっています。Agentで生成した値を設定する必要があります。

ウィジェットに値を設定する

jinja2を使うことで実現できます。以下のコードで .widgetを読み込んでtemplate部分からjinja2のテンプレートを作成し、エージェントの結果をプレースフォルダーに設定し、ウィジェットを構成します。今回使ったウィジェットはCardだったので Card.model_validate_json()としていますが、ウィジェットはListViewの場合があるのでその場合は ListView.model_validate_json()でウィジェットを構成します。

from jinja2 import Template
import json

async def respond(
    self,
    thread: ThreadMetadata,
    input: ThreadItem | None,
    context: dict[str, Any],
) -> AsyncIterator[ThreadStreamEvent]:

  # Widget Builderで作成した .widget ファイルを読み取り、jinja2のテンプレートを作成
  with open("./app/MyWidget.widget", "r", encoding="utf-8") as f:
      widget_json = json.loads(f.read())
  template = Template(widget_json["template"])

  # エージェントを実行
  result = await run_workflow(wf_input)
  data = result['output_parsed']

  # ウィジェットのプレースフォルダーにエージェントの実行結果を設定
  rendered = template.render(**data)

  # ウィジェットを構成
  card = Card.model_validate_json(rendered)

ウィジェットをアウトプットする

widgetプロパティに先ほど構成したウィジェットを設定したWidgetItemを返せばウィジェットを表示できます。

# アシスタントメッセージ 
result_item=WidgetItem(
    id = _gen_id("msg"),
    thread_id = thread.id,
    created_at = datetime.now(),  
    type="widget",
    widget=card,
)

# スレッドにアイテムを追加
yield ThreadItemAddedEvent(
    type="thread.item.added",
    item=result_item
)

# スレッドの完了
yield ThreadItemDoneEvent(item=result_item)

まとめ

Agent Builderで作成したエージェントをChatKitで実行する方法を解説しました。Agent Builderは便利ですが、それだけでは力を発揮できません。ChatKitと組み合わせることで実力を発揮します。Agent Builderに比べるとChatKitの難易度は高いですが、ChatKitの理解を深めてエージェントアプリを構築していきましょう。

この記事を書いた人