ChatKit

agent-framework-chatkit converteert OpenAI ChatKit-thread-items naar Agent Framework-berichten en zet gestreamde updates van de agent weer om in ChatKit-events. Gebruik deze als u een ChatKit-front-end wilt met een Agent Framework Python back-end.

De integratie biedt:

• ThreadItemConverter voor het converteren van threaditems van ChatKit en bijlagen.
• stream_agent_response() voor het converteren van gestreamde agentupdates naar ChatKit-gebeurtenissen.
• simple_to_agent_input() voor het standaardpad voor berichtconversie.

Prerequisites

• Python 3.10 of hoger.
• Een back-endwebframework zoals FastAPI.
• Node.js voor de chatkit-front-end.
• Een ChatKit-domeinsleutel voor een productiefront-enddomein.

Installeer het pakket

pip install agent-framework-chatkit --pre

Een ChatKit-server maken

Subklasse ChatKitServer, maak de Agent Framework-agent en configureer een conversieprogramma voor threaditems en bijlagen.

class WeatherChatKitServer(ChatKitServer[dict[str, Any]]):
    """ChatKit server implementation using 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,
        )

Antwoorden converteren en streamen

Laad de threadgeschiedenis, converteer deze naar Agent Framework-berichten, voer de agent uit in de streamingmodus en geef ChatKit-gebeurtenissen op.

async def respond(
    self,
    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.
    """
    from agent_framework import FunctionResultContent

    if input_user_message is None:
        logger.debug("Received None user message, skipping")
        return

    logger.info(f"Processing message for thread: {thread.id}")

    try:
        # Track weather data and city selector flag for this request
        weather_data: WeatherData | None = None
        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
        agent_stream = self.weather_agent.run(agent_messages, stream=True)

        # Create an intercepting stream that extracts function results while passing through updates
        async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
            nonlocal weather_data, show_city_selector
            async for update in agent_stream:
                # Check for function results in the update
                if update.contents:
                    for content in update.contents:
                        if isinstance(content, FunctionResultContent):
                            result = content.result

                            # Check if it's a WeatherResponse (string subclass with weather_data attribute)
                            if isinstance(result, str) and hasattr(result, "weather_data"):
                                extracted_data = getattr(result, "weather_data", None)
                                if isinstance(extracted_data, WeatherData):
                                    weather_data = extracted_data
                                    logger.info(f"Weather data extracted: {weather_data.location}")
                            # Check if it's the city selector marker
                            elif isinstance(result, str) and 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

ThreadItemConverter behoudt de status en vertegenwoordigt StructuredInputItem elke vraag als beantwoord, overgeslagen of onbeantwoord in een gebruikersbericht. Overschrijf de asynchrone methode structured_input_to_input() om de opmaak aan te passen of antwoorden te redigeren. Geef één Message terug, een lijst met berichten, of None om het item weg te laten.

Het conversieprogramma behoudt ook voltooide GeneratedImageItem waarden als modelcontext op latere beurten. Met de standaardconversie wordt een gebruiker Message met een korte voortekst en de gegenereerde afbeeldings-URI gemaakt. Gegevens-URI's behouden hun ingesloten mediatype, externe URL's gebruiken image/*, en onvoltooide items zonder afbeelding worden overgeslagen.

Overschrijf de asynchrone generated_image_to_input() methode om de context van de gegenereerde afbeelding aan te passen, zoals het ophalen van een privéafbeeldings-URL. Geef één Message terug, een lijst met berichten, of None om het item weg te laten.

In het volledige voorbeeld ziet u ook threads met SQLite-ondersteuning, bestandsuploads, opslag van bijlagen, acties en interactieve widgets.

Warning

De ChatKit-front-end wordt geladen vanuit het CDN van OpenAI en verzendt uitgaande aanvragen naar OpenAI-domeinen. Het kan momenteel niet zelf worden gehost en is niet geschikt voor air-gapped omgevingen.

Volgende stappen 

DevUI