ChatKit

agent-framework-chatkit OpenAI ChatKit iş parçacığı öğelerini Agent Framework iletilerine dönüştürür ve akış halindeki aracı güncellemelerini yeniden ChatKit olaylarına dönüştürür. ChatKit ön ucu ve Agent Framework Python arka ucuna sahip bir yapı istediğinizde bunu kullanın.

Tümleştirme aşağıdakileri sağlar:

  • ThreadItemConverter ChatKit ileti dizisi öğelerini ve eklerini dönüştürmek için.
  • stream_agent_response() akış hâlindeki ajan güncellemelerini ChatKit olaylarına dönüştürmek için.
  • simple_to_agent_input() varsayılan ileti dönüştürme yolu için.

Prerequisites

  • Python 3.11 veya üzeri.
  • FastAPI gibi bir arka uç web çerçevesi.
  • ChatKit ön ucu için Node.js.
  • Dağıtılmış bir modele sahip Microsoft Foundry projesi.
  • Azure CLI kimlik doğrulaması (az login).
  • Üretim ön uç etki alanı için Bir ChatKit etki alanı anahtarı.

Paketi yükle

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

Foundry modelini yapılandırın

Örnek, aracıyı AzureCliCredential ve FoundryChatClient ile oluşturur. Arka ucu başlatmadan önce proje uç noktasını ve model dağıtımını ayarlayın:

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

ChatKit sunucusu oluşturma

alt sınıfı ChatKitServer, Agent Framework aracısını oluşturun ve iş parçacığı öğeleri ve ekleri için bir dönüştürücü yapılandırın.


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

Yanıtları dönüştürme ve akışla gönderme

Konuşma geçmişini yükleyin, bunu Agent Framework mesajlarına dönüştürün, aracıyı akış modunda çalıştırın ve ChatKit olaylarını üretin.

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

Önce agent.run(..., stream=True) çağrısını yapın, ardından bu yanıt akışını ve ChatKit iş parçacığı kimliğini stream_agent_response(...) öğesine geçirin. Aracıyı dönüştürücüye geçirmeyin.

ThreadItemConverter StructuredInputItem durumu korur ve bir kullanıcı iletisinde yanıtlanan, atlanan veya yanıtlanmayan her soruyu temsil eder. Biçimlendirmeyi özelleştirmek veya yanıtları sansürlemek için asenkron structured_input_to_input() yöntemini geçersiz kılın. Bir Message, bir mesaj listesi veya öğeyi atlamak için None döndürün.

Dönüştürücü, daha sonra dönüşlerde model bağlamı olarak tamamlanmış GeneratedImageItem değerleri de korur. Varsayılan dönüştürme, kısa metin önsözü ve oluşturulan görüntü URI'sine sahip bir kullanıcı Message oluşturur. Data URI’leri gömülü medya türünü korur, harici URL’ler image/* kullanır ve resim içermeyen tamamlanmamış öğeler atlanır.

Özel görüntü URL'sini çözümleme gibi oluşturulan görüntü bağlamını özelleştirmek için zaman uyumsuz generated_image_to_input() yöntemi geçersiz kılın. Bir Message, bir mesaj listesi veya öğeyi atlamak için None döndürün.

Tam örnek ayrıca SQLite destekli iş parçacıklarını, dosya yüklemelerini, ek depolama alanını, eylemleri ve etkileşimli pencere öğelerini gösterir.

Warning

ChatKit ön ucu OpenAI'nin CDN'sinden yüklenir ve OpenAI etki alanlarına giden istekler yapar. Şu anda kendi sunucularınızda barındırılamaz ve ağdan fiziksel olarak izole ortamlara uygun değildir.

Sonraki Adımlar

DevUI

  • Son güncelleştirme: