ChatKit

agent-framework-chatkit konvertiert OpenAI ChatKit-Threadelemente in Agent Framework-Nachrichten und konvertiert gestreamte Agent-Updates wieder in ChatKit-Ereignisse. Verwenden Sie es, wenn Sie ein ChatKit-Frontend mit einem Agent Framework Python Back-End verwenden möchten.

Die Integration bietet Folgendes:

• ThreadItemConverter zum Konvertieren von Elementen eines ChatKit-Threads und Anhängen.
• stream_agent_response() zum Konvertieren von Stream-Agent-Updates in ChatKit-Ereignisse.
• simple_to_agent_input() für den Standardmäßigen Nachrichtenkonvertierungspfad.

Voraussetzungen

• Python 3.11 oder höher.
• Ein Back-End-Webframework wie FastAPI.
• Node.js für das ChatKit-Frontend.
• Ein ChatKit-Domänenschlüssel für eine Front-End-Produktionsdomäne.

Installiere das Paket

pip install agent-framework-chatkit --pre

Erstellen eines ChatKit-Servers

Unterklasse ChatKitServer, erstellen Sie den Agent Framework-Agent, und konfigurieren Sie einen Konverter für Threadelemente und Anlagen.

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

Konvertieren und Antworten streamen

Laden Sie den Threadverlauf, konvertieren Sie ihn in Agent Framework-Nachrichten, führen den Agent im Streamingmodus aus und liefern ChatKit-Ereignisse.

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

Rufen Sie zuerst agent.run(..., stream=True) auf und übergeben Sie dann diesen Antwortdatenstrom und die ChatKit-Thread-ID an stream_agent_response(...). Übergeben Sie den Agenten selbst nicht an den Konverter.

ThreadItemConverter StructuredInputItem behält den Status bei und stellt jede Frage als beantwortet, übersprungen oder nicht beantwortet in einer Benutzernachricht dar. Überschreiben Sie die asynchrone Methode structured_input_to_input(), um die Formatierung anzupassen oder Antworten zu schwärzen. Geben Sie ein Message, eine Liste von Nachrichten oder None zurück, um das Element auszulassen.

Der Konverter bewahrt auch vervollständigte GeneratedImageItem-Werte in späteren Interaktionen als Modellkontext auf. Die Standardkonvertierung erstellt eine Benutzer-Message mit einer kurzen vorangestellten Text und dem generierten Bild-URI. Daten-URIs behalten ihren eingebetteten Medientyp, externe URLs verwenden image/*und nicht fertige Elemente ohne Bild werden übersprungen.

Überschreiben Sie die asynchrone Methode generated_image_to_input(), um den generierten Bildkontext anzupassen, z. B. indem eine private Bild-URL aufgelöst wird. Geben Sie ein Message, eine Liste von Nachrichten oder None zurück, um das Element auszulassen.

Das vollständige Beispiel veranschaulicht außerdem SQLite-gesicherte Threads, Dateiuploads, Anlagenspeicher, Aktionen und interaktive Widgets.

Warnung

Das ChatKit-Frontend wird aus dem OpenAI CDN geladen und sendet ausgehende Anforderungen an OpenAI-Domänen. Es kann derzeit nicht selbst gehostet werden und eignet sich nicht für luftgespaltene Umgebungen.

Nächste Schritte

DevUI