ChatKit

agent-framework-chatkit převede položky vlákna OpenAI ChatKit na zprávy rozhraní Agent Framework a převede streamované aktualizace agenta zpět na události ChatKit. Použijte ho, pokud chcete frontend ChatKitu a backend v Pythonu využívající Agent Framework.

Integrace poskytuje:

• ThreadItemConverter pro převod položek vlákna a příloh ChatKit.
• stream_agent_response() pro převod streamovaných aktualizací agentů na události ChatKitu.
• simple_to_agent_input() pro výchozí cestu převodu zpráv.

Prerequisites

• Python 3.11 nebo novější.
• Back-endová webová architektura, jako je FastAPI.
• Node.js pro frontend ChatKit.
• Projekt Microsoft Foundry s nasazeným modelem.
• ověřování Azure CLI (az login).
• Klíč domény ChatKit pro produkční front-endovou doménu.

Nainstalujte balíček

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

Nakonfigurujte model Foundry

Ukázka vytvoří agenta s FoundryChatClient a AzureCliCredential. Před spuštěním back-endu nastavte koncový bod projektu a nasazení modelu:

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

Vytvoření serveru ChatKit

Podtřídu ChatKitServer, vytvořte agenta Agent Framework a nakonfigurujte převaděč pro položky vlákna a přílohy.

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

Převádějte odpovědi a streamujte je

Načtěte historii vlákna, převeďte ji do podoby zpráv frameworku Agent Framework, spusťte agenta ve streamovacím režimu a vracejte události 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):

Nejprve zavolejte agent.run(..., stream=True), pak předejte tento stream odpovědi a ID vlákna ChatKitu do stream_agent_response(...). Nepředávejte samotného agenta konvertoru.

ThreadItemConverter StructuredInputItem zachovává stav a představuje každou otázku jako zodpovězenou, přeskočenou nebo nezodpovězenou ve zprávě uživatele. Přepište asynchronní metodu structured_input_to_input(), abyste mohli přizpůsobit formátování nebo redigovat odpovědi. Vraťte jednu Message, seznam zpráv nebo položku vynechejte pomocí None.

Konvertor také zachovává dokončené hodnoty GeneratedImageItem jako kontext modelu v dalších kolech. Výchozí konverze vytvoří uživatele Message s krátkým úvodním textem a vygenerovaným URI obrázku. Datové identifikátory URI si zachovávají vložený typ média, externí adresy URL používají image/* a neúplné položky bez obrázku se přeskočí.

Přepište asynchronní metodu generated_image_to_input() a přizpůsobte kontext vygenerovaného obrázku, například při zpracování privátní adresy URL obrázku. Vraťte jednu Message, seznam zpráv nebo položku vynechejte pomocí None.

Kompletní ukázka také ukazuje vlákna s podporou SQLite, nahrávání souborů, úložiště příloh, akce a interaktivní widgety.

Warning

Front-end ChatKitu se načte z CDN OpenAI a provádí odchozí požadavky na domény OpenAI. V současné době ji nelze provozovat on-premise a není vhodná pro izolovaná prostředí.

Další kroky

DevUI