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 front-end ChatKitu s architekturou agentů Python back-end.

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.

Předpoklady

• Python 3.10 nebo novější.
• Back-endová webová architektura, jako je FastAPI.
• Node.js pro front-end ChatKit.
• Klíč domény ChatKit pro produkční front-endovou doménu.

Nainstalujte balíček

pip install agent-framework-chatkit --pre

Vytvoření serveru ChatKit

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

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

Převod odpovědí na stream a převod odpovědí

Načtěte historii vláken, převeďte ji na zprávy rozhraní Agent Framework, spusťte agenta v režimu streamování a výnos událostí ChatKit.

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

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ě není možné ji hostovat v místním prostředí a není vhodná pro prostředí s mezerami vzduchu.

Další kroky

DevUI