ChatKit

agent-framework-chatkit convierte los elementos del hilo de ChatKit de OpenAI en mensajes de Agent Framework y convierte las actualizaciones del agente transmitidas en streaming de nuevo en eventos de ChatKit. Úselo cuando desee una interfaz de ChatKit con un backend en Python de Agent Framework.

La integración proporciona:

• ThreadItemConverter para convertir elementos de hilo y archivos adjuntos de ChatKit.
• stream_agent_response() para convertir las actualizaciones transmitidas del agente en eventos de ChatKit.
• simple_to_agent_input() para la ruta de conversión de mensajes predeterminada.

Prerequisites

• Python 3.11 o posterior.
• Un marco web back-end, como FastAPI.
• Node.js para el front-end de ChatKit.
• Un proyecto de Microsoft Foundry con un modelo implementado.
• Autenticación de CLI de Azure (az login).
• Una clave de dominio de ChatKit para un dominio de front-end de producción.

Instalar el paquete

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

Configurar el modelo Foundry

En el ejemplo se crea el agente con FoundryChatClient y AzureCliCredential. Establezca el punto de conexión del proyecto y la implementación del modelo antes de iniciar el back-end:

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

Creación de un servidor de ChatKit

Cree una subclase de ChatKitServer, cree un agente de Agent Framework y configure un convertidor para los elementos del hilo y los archivos adjuntos.

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

Conversión y transmisión de respuestas

Cargue el historial de subprocesos, conviértelo en mensajes de Agent Framework, ejecute el agente en modo de streaming y produzca eventos de 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):

Llame primero a agent.run(..., stream=True) y, a continuación, pase esa secuencia de respuesta y el identificador de subproceso de ChatKit a stream_agent_response(...). No pases el propio agente al convertidor.

ThreadItemConverter conserva el StructuredInputItem estado y representa cada pregunta como contestada, omitida o sin responder en un mensaje de usuario. Invalide el método asincrónico structured_input_to_input() para personalizar el formato o censurar respuestas. Devuelve un Message, una lista de mensajes o None para omitir el elemento.

El convertidor también conserva los valores completados GeneratedImageItem como contexto de modelo en turnos posteriores. La conversión predeterminada crea un usuario Message con un prefacio de texto corto y el URI de imagen generado. Las URI de datos conservan su tipo de medio incrustado, las URL externas usan image/*, y se omiten los elementos incompletos sin imagen.

Invalide el método asincrónico generated_image_to_input() para personalizar el contexto de imagen generado, como resolver una dirección URL de imagen privada. Devuelve un Message, una lista de mensajes o None para omitir el elemento.

En el ejemplo completo también se muestran subprocesos respaldados por SQLite, cargas de archivos, almacenamiento de datos adjuntos, acciones y widgets interactivos.

Warning

El front-end de ChatKit se carga desde la red CDN de OpenAI y realiza solicitudes salientes a dominios de OpenAI. Actualmente no puede ser autohospedado y no es adecuado para entornos con disponibilidad de aire.

Pasos siguientes

DevUI