ChatKit

agent-framework-chatkit mengonversi item thread OpenAI ChatKit menjadi pesan Agent Framework dan mengonversi pembaruan agen yang dialirkan secara streaming menjadi peristiwa ChatKit. Gunakan saat Anda menginginkan frontend ChatKit dengan Agent Framework Python backend.

Integrasi ini menyediakan:

• ThreadItemConverter untuk mengonversi item-item dan lampiran thread ChatKit.
• stream_agent_response() untuk mengonversi pembaruan dari agen yang di-streaming menjadi event ChatKit.
• simple_to_agent_input() untuk jalur konversi pesan default.

Prerequisites

• Python 3.11 atau yang lebih baru.
• Kerangka kerja web backend seperti FastAPI.
• Node.js untuk frontend ChatKit.
• Proyek Microsoft Foundry dengan model yang diterapkan.
• autentikasi Azure CLI (az login).
• Kunci domain ChatKit untuk domain frontend produksi.

Pasang paketnya

pip install agent-framework-chatkit agent-framework-foundry --pre

Konfigurasikan model Foundry

Sampel membuat agen dengan FoundryChatClient dan AzureCliCredential. Atur titik akhir proyek dan deployment model sebelum Anda memulai backend:

export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL="<model-deployment>"

Membuat server ChatKit

Buat subkelas dari ChatKitServer, buat agen Agent Framework, dan konfigurasikan pengonversi untuk item utas dan lampiran.

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")

Mengonversi dan mengalirkan respons

Muat riwayat utas, konversikan menjadi pesan Agent Framework, jalankan agen dalam mode streaming, dan hasilkan peristiwa 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):

Hubungi agent.run(..., stream=True) terlebih dahulu, lalu teruskan aliran respons tersebut dan ID utas ChatKit ke stream_agent_response(...). Jangan meneruskan agen itu sendiri ke pengonversi.

ThreadItemConverter StructuredInputItem mempertahankan status dan mewakili setiap pertanyaan sebagai jawaban, dilewati, atau tidak terjawab dalam pesan pengguna. Ambil alih metode asinkron structured_input_to_input() untuk menyesuaikan pemformatan atau meredaksi jawaban. Kembalikan satu Message, daftar pesan, atau None untuk mengabaikan item.

Konverter juga mempertahankan nilai yang sudah diisi GeneratedImageItem sebagai konteks model pada giliran berikutnya. Konversi default membuat user Message dengan teks pengantar singkat dan URI gambar yang dihasilkan. URI data mempertahankan jenis media yang disematkan, URL eksternal menggunakan image/*, dan item yang belum selesai tanpa gambar dilewati.

Ambil alih metode asinkron generated_image_to_input() untuk menyesuaikan konteks gambar yang dihasilkan, seperti menyelesaikan URL gambar privat. Kembalikan satu Message, daftar pesan, atau None untuk mengabaikan item.

Sampel lengkap juga menunjukkan utas yang didukung SQLite, unggahan file, penyimpanan lampiran, tindakan, dan widget interaktif.

Warning

Frontend ChatKit dimuat dari CDN OpenAI dan membuat permintaan keluar ke domain OpenAI. Saat ini tidak dapat dihosting sendiri dan tidak cocok untuk lingkungan air-gapped.

Langkah berikutnya

DevUI