ChatKit

agent-framework-chatkit تحويل عناصر مؤشر ترابط OpenAI ChatKit إلى رسائل إطار عمل العامل وتحويل تحديثات العامل المتدفقة مرة أخرى إلى أحداث ChatKit. استخدمه عندما تريد واجهة ChatKit أمامية مع إطار عمل عامل Python الخلفية.

يوفر التكامل ما يلي:

• ThreadItemConverter لتحويل عناصر ومرفقات مؤشر ترابط ChatKit.
• stream_agent_response() لتحويل تحديثات العامل المتدفقة إلى أحداث ChatKit.
• simple_to_agent_input() لمسار تحويل الرسائل الافتراضي.

المتطلبات الأساسية

• Python 3.11 أو أحدث.
• إطار عمل ويب خلفي مثل FastAPI.
• Node.js للواجهة الأمامية ChatKit.
• مشروع Microsoft Foundry مع نموذج منشور.
• مصادقة Azure CLI (az login).
• مفتاح مجال ChatKit لمجال الواجهة الأمامية للإنتاج.

تثبيت الحزمة

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

تكوين نموذج Foundry

ينشئ النموذج العامل مع FoundryChatClient و AzureCliCredential. تعيين نقطة نهاية المشروع ونشر النموذج قبل بدء تشغيل الخلفية:

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

إنشاء خادم ChatKit

الفئة ChatKitServerالفرعية ، أنشئ عامل إطار عمل العامل، وقم بتكوين محول لعناصر مؤشر الترابط والمرفقات.

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

تحويل الاستجابات ودفقها

قم بتحميل محفوظات مؤشر الترابط، وتحويله إلى رسائل إطار عمل العامل، وتشغيل العامل في وضع الدفق، وإخراج أحداث 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):

اتصل agent.run(..., stream=True) أولا، ثم مرر دفق الاستجابة هذا ومعرف مؤشر ترابط ChatKit إلى stream_agent_response(...). لا تمرر العامل نفسه إلى المحول.

ThreadItemConverter يحافظ StructuredInputItem على الحالة ويمثل كل سؤال على أنه مجاب عنه أو تم تخطيه أو غير مجاب عنه في رسالة مستخدم. تجاوز الأسلوب غير structured_input_to_input() المتزامن لتخصيص التنسيق أو تنقيح الإجابات. إرجاع واحد Messageأو قائمة بالرسائل أو None لحذف العنصر.

يحافظ المحول أيضا على القيم المكتملة GeneratedImageItem بسياق نموذج عند التشغيل اللاحق. ينشئ التحويل الافتراضي مستخدما Message مزودا بمقدمة نص قصيرة وURI للصورة التي تم إنشاؤها. تحتفظ معرفات URI للبيانات بنوع الوسائط المضمنة، وتستخدم عناوين URL image/*الخارجية ، ويتم تخطي العناصر غير المنتهية بدون صورة.

تجاوز الأسلوب غير المتزامن generated_image_to_input() لتخصيص سياق الصورة الذي تم إنشاؤه، مثل حل عنوان URL للصورة الخاصة. إرجاع واحد Messageأو قائمة بالرسائل أو None لحذف العنصر.

يوضح النموذج الكامل أيضا مؤشرات الترابط المدعومة من SQLite وتحميلات الملفات وتخزين المرفقات والإجراءات وعناصر واجهة المستخدم التفاعلية.

تحذير

يتم تحميل الواجهة الأمامية ChatKit من شبكة تسليم المحتوى الخاصة ب OpenAI وتقدم طلبات صادرة إلى مجالات OpenAI. لا يمكن أن تكون مستضافة ذاتيا حاليا وغير مناسبة للبيئات المكيفة الهواء.

الخطوات التالية

واجهة مستخدم DevUI