ChatKit

agent-framework-chatkit Konwertuje elementy wątku OpenAI ChatKit na komunikaty struktury agentów i konwertuje przesyłane strumieniowo aktualizacje agenta z powrotem na zdarzenia ChatKit. Użyj tego, gdy chcesz mieć frontend ChatKit z backendem Agent Framework napisanym w Pythonie.

Integracja zapewnia:

• ThreadItemConverter do konwersji elementów wątku ChatKit i załączników.
• stream_agent_response() do konwertowania aktualizacji agenta przesyłanego strumieniowo na zdarzenia ChatKit.
• simple_to_agent_input() dla domyślnej ścieżki konwersji komunikatów.

Wymagania wstępne

• Python 3.10 lub nowszy.
• Backendowy framework webowy, taki jak FastAPI.
• Node.js dla frontendu ChatKit.
• Klucz domeny ChatKit dla domeny frontonu produkcyjnego.

Instalowanie pakietu

pip install agent-framework-chatkit --pre

Tworzenie serwera ChatKit

Podklasa ChatKitServer, utwórz agenta platformy Agent Framework i skonfiguruj konwerter dla elementów wątków i załączników.

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

Konwertować i strumieniować odpowiedzi

Załaduj historię wątków, przekonwertuj ją na komunikaty platformy Agent, uruchom agenta w trybie przesyłania strumieniowego i wylicz zdarzenia 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

ThreadItemConverter zachowuje stan StructuredInputItem i przedstawia każde pytanie w wiadomości użytkownika jako takie, na które udzielono odpowiedzi, pominięte lub bez odpowiedzi. Przesłoń asynchroniczną metodę structured_input_to_input(), aby dostosować formatowanie lub redagować odpowiedzi. Zwróć jeden Message, listę komunikatów lub None, aby pominąć element.

Konwerter zachowuje również uzupełnione wartości GeneratedImageItem jako kontekst modelu w kolejnych turach. Konwersja domyślna tworzy użytkownika Message z krótkim prefiksem tekstowym i wygenerowanym identyfikatorem URI obrazu. URI danych zachowują swój osadzony typ MIME, zewnętrzne adresy URL używają image/*, a niedokończone elementy bez obrazu są pomijane.

Przesłoń metodę asynchroniczną generated_image_to_input(), aby dostosować wygenerowany kontekst obrazu, na przykład ustalając prywatny adres URL obrazu. Zwróć jeden Message, listę komunikatów lub None, aby pominąć element.

Kompletny przykład przedstawia również wątki oparte na platformie SQLite, przekazywanie plików, magazyn załączników, akcje i interaktywne widżety.

Warning

Fronton ChatKit jest ładowany z usługi CDN platformy OpenAI i wysyła żądania wychodzące do domen OpenAI. Obecnie nie można tego wdrożyć lokalnie i nie nadaje się to do środowisk odizolowanych od sieci.

Następne kroki

DevUI