ChatKit

agent-framework-chatkit converte itens do thread OpenAI ChatKit em mensagens do Agent Framework e converte as atualizações transmitidas de agentes de volta em eventos ChatKit. Usa-o quando quiseres um frontend ChatKit com um backend Python do Agent Framework.

A integração proporciona:

• ThreadItemConverter para converter itens e anexos do tópico do ChatKit.
• stream_agent_response() para converter atualizações de agentes transmitidas em eventos do ChatKit.
• simple_to_agent_input() para o caminho padrão de conversão de mensagens.

Pré-requisitos

• Python 3.10 ou posterior.
• Um framework web backend como o FastAPI.
• Node.js para o frontend do ChatKit.
• Uma chave de domínio ChatKit para um domínio frontend de produção.

Instale o pacote

pip install agent-framework-chatkit --pre

Criar um servidor ChatKit

Subclasse ChatKitServer, crie o agente Agent Framework e configure um conversor para itens de thread e anexos.

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

Conversão e fluxo de respostas

Carrega o histórico da thread, converte-o para mensagens do Agent Framework, executa o agente em modo streaming e gera eventos do 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

O exemplo completo também demonstra threads suportados por SQLite, carregamentos de ficheiros, armazenamento de anexos, ações e widgets interativos.

Warning

A interface ChatKit é carregada a partir da CDN da OpenAI e faz pedidos de saída para domínios OpenAI. Neste momento, não pode ser auto-hospedado e não é adequado para ambientes isolados.

Passos seguintes

DevUI