agent-framework-chatkit は、OpenAI ChatKit スレッド項目を Agent Framework メッセージに変換し、ストリーミングされたエージェントの更新を ChatKit イベントに変換します。 Agent Framework Python バックエンドで ChatKit フロントエンドを使用する場合に使用します。
統合により、次の機能が提供されます。
-
ThreadItemConverterChatKit スレッド項目と添付ファイルを変換する場合。 -
stream_agent_response()ストリーミング されたエージェントの更新を ChatKit イベントに変換する場合に使用します。 -
simple_to_agent_input()既定のメッセージ変換パス。
前提条件
- Python 3.11 以降。
- FastAPI などのバックエンド Web フレームワーク。
- ChatKit フロントエンドの Node.js。
- デプロイされたモデルを含む Microsoft Foundry プロジェクト。
- Azure CLI認証 (
az login)。 - 運用フロントエンド ドメインの ChatKit ドメイン キー。
パッケージをインストールする
pip install agent-framework-chatkit agent-framework-foundry --pre
Foundry モデルを構成する
このサンプルでは、 FoundryChatClient と AzureCliCredentialを使用してエージェントを作成します。
バックエンドを開始する前に、プロジェクト エンドポイントとモデルのデプロイを設定します。
export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL="<model-deployment>"
ChatKit サーバーを作成する
ChatKitServerサブクラス化し、Agent Framework エージェントを作成し、スレッド項目と添付ファイルのコンバーターを構成します。
This server integrates Agent Framework agents with ChatKit's server protocol,
providing weather information with interactive widgets and time queries through Azure OpenAI.
"""
def __init__(self, data_store: SQLiteStore, attachment_store: FileBasedAttachmentStore):
super().__init__(data_store, attachment_store)
logger.info("Initializing WeatherChatKitServer")
# Create Agent Framework agent with Azure OpenAI
# For authentication, run `az login` command in terminal
try:
self.weather_agent = Agent(
client=FoundryChatClient(credential=AzureCliCredential()),
instructions=(
"You are a helpful weather assistant with image analysis capabilities. "
"You can provide weather information for any location, tell the current time, "
"and analyze images that users upload. Be friendly and informative in your responses.\n\n"
"If a user asks to see a list of cities or wants to choose from available cities, "
"use the show_city_selector tool to display an interactive city selector.\n\n"
"When users upload images, you will automatically receive them and can analyze their content. "
"Describe what you see in detail and be helpful in answering questions about the images."
),
tools=[get_weather, get_time, show_city_selector],
)
logger.info("Weather agent initialized successfully with Azure OpenAI")
except Exception as e:
logger.error(f"Failed to initialize weather agent: {e}")
raise
# Create ThreadItemConverter with attachment data fetcher
self.converter = ThreadItemConverter(
attachment_data_fetcher=self._fetch_attachment_data,
)
logger.info("WeatherChatKitServer initialized")
応答の変換とストリーム配信
スレッド履歴を読み込み、Agent Framework メッセージに変換し、ストリーミング モードでエージェントを実行し、ChatKit イベントを生成します。
thread: ThreadMetadata,
input_user_message: UserMessageItem | None,
context: dict[str, Any],
) -> AsyncIterator[ThreadStreamEvent]:
"""Handle incoming user messages and generate responses.
This method converts ChatKit messages to Agent Framework format using ThreadItemConverter,
runs the agent, converts the response back to ChatKit events using stream_agent_response,
and creates interactive weather widgets when weather data is queried.
"""
if input_user_message is None:
logger.debug("Received None user message, skipping")
return
logger.info(f"Processing message for thread: {thread.id}")
try:
# Collect weather data from get_weather and track the city selector flag for this request
weather_results: list[WeatherData] = []
show_city_selector = False
# Load full thread history from the store
thread_items_page = await self.store.load_thread_items(
thread_id=thread.id,
after=None,
limit=1000,
order="asc",
context=context,
)
thread_items = thread_items_page.data
# Convert ALL thread items to Agent Framework ChatMessages using ThreadItemConverter
# This ensures the agent has the full conversation context
agent_messages = await self.converter.to_agent_input(thread_items)
if not agent_messages:
logger.warning("No messages after conversion")
return
logger.info(f"Running agent with {len(agent_messages)} message(s)")
# Run the Agent Framework agent with streaming. get_weather appends to weather_results.
agent_stream = self.weather_agent.run(
agent_messages,
stream=True,
function_invocation_kwargs={"weather_results": weather_results},
)
# Create an intercepting stream that detects the city selector marker while passing through updates
async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
nonlocal show_city_selector
async for update in agent_stream:
# Check for the city selector marker in function results
for content in update.contents:
if content.type == "function_result" and content.result == "__SHOW_CITY_SELECTOR__":
show_city_selector = True
logger.info("City selector flag detected")
yield update
# Stream updates as ChatKit events with interception
async for event in stream_agent_response(
intercept_stream(),
thread_id=thread.id,
):
yield event
# If weather data was collected during the tool call, create a widget
if weather_results:
weather_data = weather_results[-1]
logger.info(f"Creating weather widget for location: {weather_data.location}")
# Create weather widget
widget = render_weather_widget(weather_data)
copy_text = weather_widget_copy_text(weather_data)
# Stream the widget
async for widget_event in stream_widget(thread_id=thread.id, widget=widget, copy_text=copy_text):
最初に agent.run(..., stream=True) を呼び出し、その応答ストリームと ChatKit スレッド ID を stream_agent_response(...) に渡します。 エージェント自体をコンバーターに渡さないでください。
ThreadItemConverter は StructuredInputItem 状態を保持し、各質問をユーザー メッセージで回答、スキップ、または未回答として表します。 非同期 structured_input_to_input() メソッドをオーバーライドして、書式設定をカスタマイズしたり、回答を編集したりします。 1 つの Message、メッセージの一覧、または項目を省略する None を返します。
コンバーターは、後のターンで完了した GeneratedImageItem 値をモデル コンテキストとして保持します。 既定の変換では、短いテキストの前書きと生成されたイメージ URI を持つユーザー Message が作成されます。 データ URI は埋め込みメディアの種類を保持し、外部 URL は image/*を使用し、画像のない未完成のアイテムはスキップされます。
プライベート イメージ URL の解決など、生成されたイメージ コンテキストをカスタマイズするには、非同期の generated_image_to_input() メソッドをオーバーライドします。 1 つの Message、メッセージの一覧、または項目を省略する None を返します。
完全なサンプルでは、SQLite ベースのスレッド、ファイルのアップロード、添付ファイルストレージ、アクション、対話型ウィジェットも示しています。
Warning
ChatKit フロントエンドは OpenAI の CDN から読み込まれ、OpenAI ドメインへの送信要求を行います。 現在、セルフホステッドにすることはできないため、エアギャップ環境には適していません。