Azure异步流场景下Assistant函数调用结果传递失败求助

在Azure异步流示例中实现向Assistant传递函数时遇到问题：函数调用结果已在日志中显示，但Assistant返回错误信息：

{"function_error": "find_and_get_doctors_info", "error": "Function is not available."}

提示未接收结果且函数当前不可用。已在Azure Studio完成函数描述配置，尝试直接将结果发送至异步消息队列仍无效。

主代码 #

import asyncio
import json
import os
import random

from azure.ai.assistant.management.async_assistant_client import AsyncAssistantClient
from azure.ai.assistant.management.ai_client_factory import AsyncAIClientType
from azure.ai.assistant.management.async_assistant_client_callbacks import AsyncAssistantClientCallbacks
from azure.ai.assistant.management.async_conversation_thread_client import AsyncConversationThreadClient
from azure.ai.assistant.management.async_message import AsyncConversationMessage
from azure.ai.assistant.management.text_message import TextMessage
from services.gpt_service.tools import available_functions, tools_any_b2c  # Import the available functions
from configs.openai.assistant_config import PHRASES
from services import tts_service

conversation_threads = {}

class MyAssistantClientCallbacks(AsyncAssistantClientCallbacks):
    def __init__(self, message_queue):
        self.message_queue = message_queue

    async def handle_message(self, action, message="hello"):
        await self.message_queue.put((action, message))

    async def on_run_update(self, assistant_name, run_identifier, run_status, thread_name, is_first_message=False,
                            message: AsyncConversationMessage = None):
        if run_status == "streaming":
            await self.handle_message("start" if is_first_message else "message", message.text_message.content)
        elif run_status == "completed":
            if message:
                text_message: TextMessage = message.text_message
                if text_message.file_citations:
                    for file_citation in text_message.file_citations:
                        print(
                            f"\nFile citation, file_id: {file_citation.file_id}, file_name: {file_citation.file_name}")

    async def on_function_call_processed(self, assistant_name, run_identifier, function_name, arguments, response):
        await self.handle_message("function", function_name)

        if function_name in available_functions:
            try:
                args = json.loads(arguments)

                func = available_functions[function_name]
                result = await func(**args)

                await self.handle_message("function_result", json.dumps(result))
            except Exception as e:
                error_message = f"Error calling function {function_name}: {str(e)}"
                await self.handle_message("function_error", error_message)
        else:
            await self.handle_message("function_error", f"Unknown function: {function_name}")

async def display_streamed_messages(message_queue, assistant_name):
    while True:
        message_type, message = await message_queue.get()
        if message_type == "start":
            print(f"{assistant_name}: {message}", end="")
        elif message_type == "message":
            print(message, end="", flush=True)
        elif message_type == "function":
            print(f"{assistant_name}: called {message} function.")
        elif message_type == "function_result":
            print(f"Function result: {message}")
        elif message_type == "function_error":
            print(f"Function error: {message}")
        message_queue.task_done()

async def gpt_main(user_message, user_id):
    assistant_name = "b2c_camny"

    try:
        with open(f"configs/azure_config/assistants/{assistant_name}_assistant_config.json", "r") as file:
            config = file.read()
    except FileNotFoundError:
        print(f"Configuration file for {assistant_name} not found.")
        return

    try:
        message_queue = asyncio.Queue()
        callbacks = MyAssistantClientCallbacks(message_queue)

        assistant_client = await AsyncAssistantClient.from_json(config, callbacks=callbacks,
                                                                api_key=os.environ["AZURE_OPENAI_API_KEY"])
        ai_client_type = AsyncAIClientType[assistant_client.assistant_config.ai_client_type]

        conversation_thread_client = AsyncConversationThreadClient.get_instance(ai_client_type)

        if user_id in conversation_threads:
            thread_name = conversation_threads[user_id]
            print(f"Reusing existing thread {thread_name} for user {user_id}")
        else:
            thread_name = await conversation_thread_client.create_conversation_thread()
            if not thread_name:
                print(f"Failed to create conversation thread for user {user_id}")
                return
            conversation_threads[user_id] = thread_name
            print(f"Created new thread {thread_name} for user {user_id}")

        await conversation_thread_client.create_conversation_thread_message(user_message, thread_name)

        print(f"Starting to process messages for thread: {thread_name}")
        await assistant_client.process_messages(thread_name=thread_name, stream=True)

        while True:
            try:
                message_type, message = await asyncio.wait_for(message_queue.get(), timeout=60)
                if message_type in ("start", "message"):
                    print(f"Message received: {message}")
                    yield message
                elif message_type == "function_result":
                    print(f"Function result received: {message}")
                    await MyAssistantClientCallbacks.on_function_call_processed()
                    await asyncio.create_task(display_streamed_messages(message_queue, assistant_name))
                elif message_type == "function_error":
                    print(f"Function error received: {message}")
                message_queue.task_done()
            except asyncio.TimeoutError:
                print(f"No messages received within the timeout period.")
                break

    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        print(f"Finished processing for thread: {thread_name}")


async def handle_message(user_message, user_id):
    async for response in gpt_main(user_message, user_id):
        print(response)

工具函数代码 #

from typing import List, Dict, Any

async def find_and_get_doctors_info(date: str, service: str, city: str) -> List[Dict[str, Any]]:
    print(f"ChatGPT call tool: find_and_get_doctors_info, date: {date}, service: {service}, city: {city}")

    if service == "Cardiologist":
        return [{
            "doctor_id": "12345",
            "full_name": "Dr. Olena Petrova",
            "doctor_type": "Cardiologist",
            "available_time": "10:00-14:00"
        }]
    elif service == "Dermatologist":
        return [{
            "doctor_id": "54321",
            "full_name": "Dr. Mykhaylo Ivanov",
            "doctor_type": "Dermatologist",
            "available_time": "13:00-18:00"
        }]
    elif service == "Dentist":
        return [{
            "doctor_id": "67890",
            "full_name": "Dr. Sofia Kovalenko",
            "doctor_type": "Dentist",
            "available_time": "09:00-12:00"
        }]
    elif service == "Ophthalmologist":
        return [{
            "doctor_id": "21436",
            "full_name": "Dr. Dmytro Bondarenko",
            "doctor_type": "Ophthalmologist",
            "available_time": "15:00-20:00"
        }]
    elif service == "Pediatrician":
        return [{
            "doctor_id": "85274",
            "full_name": "Dr. Anna Sidorenko",
            "doctor_type": "Pediatrician",
            "available_time": "11:00-16:00"
        }]
    elif service == "Neurologist":
        return [{
            "doctor_id": "93715",
            "full_name": "Dr. Andriy Shevchenko",
            "doctor_type": "Neurologist",
            "available_time": "14:00-19:00",
            "clinic_name": "Neuro Clinic",
        }]
    elif service == "Pulmonologist":
        return [{
            "doctor_id": "70192",
            "full_name": "Dr. Iryna Popel",
            "doctor_type": "Pulmonologist",
            "available_time": "12:00-17:00"
        }]
    elif service == "Gastroenterologist":
        return [{
            "doctor_id": "38651",
            "full_name": "Dr. Bohdan Kravchenko",
            "doctor_type": "Gastroenterologist",
            "available_time": "10:30-15:30",
            "clinic_name": "Gastro Clinic",
        }]
    elif service == "Orthopedist":
        return [{
            "doctor_id": "49573",
            "full_name": "Dr. Yulia Petrova",
            "doctor_type": "Orthopedist",
            "available_time": "08:00-13:00"
        }]
    elif service == "Urologist":
        return [{
            "doctor_id": "01289",
            "full_name": "Dr. Oleksandr Ivanov",
            "doctor_type": "Urologist",
            "available_time": "16:00-21:00"
        }]
    else:
        return [{ "result": "No doctors available for the selected service and date" }]

available_functions = {
    "find_and_get_doctors_info": find_and_get_doctors_info
}

核心问题 #

代码仅将函数结果发送到消息队列，未将结果提交回Assistant的Run上下文，导致Assistant无法获取返回值，进而判定函数不可用。

具体修复步骤 #

1. 修正on_function_call_processed方法 #

该方法需要将函数执行结果或错误返回给AsyncAssistantClient，而非仅发送到消息队列：

async def on_function_call_processed(self, assistant_name, run_identifier, function_name, arguments, response):
    await self.handle_message("function", function_name)
    result = None
    error = None
    if function_name in available_functions:
        try:
            args = json.loads(arguments)
            func = available_functions[function_name]
            result = await func(**args)
            await self.handle_message("function_result", json.dumps(result))
        except Exception as e:
            error = f"Error calling function {function_name}: {str(e)}"
            await self.handle_message("function_error", error)
    else:
        error = f"Unknown function: {function_name}"
        await self.handle_message("function_error", error)
    # 关键：将结果/错误返回给Assistant以继续Run流程
    return result, error

2. 移除无效的函数调用 #

在gpt_main的消息处理逻辑中，删除以下两行无效代码：

await MyAssistantClientCallbacks.on_function_call_processed()
await asyncio.create_task(display_streamed_messages(message_queue, assistant_name))

3. 验证函数名称一致性 #

确保Azure Studio中配置的函数名称与代码中的find_and_get_doctors_info完全一致，包括大小写、下划线等细节。

4. 确认工具函数注册 #

检查available_functions字典是否正确导入，且find_and_get_doctors_info函数被正确映射。

修复后关键代码片段 #

修正后的Callbacks类 #

class MyAssistantClientCallbacks(AsyncAssistantClientCallbacks):
    def __init__(self, message_queue):
        self.message_queue = message_queue

    async def handle_message(self, action, message="hello"):
        await self.message_queue.put((action, message))

    async def on_run_update(self, assistant_name, run_identifier, run_status, thread_name, is_first_message=False,
                            message: AsyncConversationMessage = None):
        if run_status == "streaming":
            await self.handle_message("start" if is_first_message else "message", message.text_message.content)
        elif run_status == "completed":
            if message:
                text_message: TextMessage = message.text_message
                if text_message.file_citations:
                    for file_citation in text_message.file_citations:
                        print(
                            f"\nFile citation, file_id: {file_citation.file_id}, file_name: {file_citation.file_name}")

    async def on_function_call_processed(self, assistant_name, run_identifier, function_name, arguments, response):
        await self.handle_message("function", function_name)
        result = None
        error = None
        if function_name in available_functions:
            try:
                args = json.loads(arguments)
                func = available_functions[function_name]
                result = await func(**args)
                await self.handle_message("function_result", json.dumps(result))
            except Exception as e:
                error = f"Error calling function {function_name}: {str(e)}"
                await self.handle_message("function_error", error)
        else:
            error = f"Unknown function: {function_name}"
            await self.handle_message("function_error", error)
        return result, error

修正后的消息处理逻辑 #

while True:
    try:
        message_type, message = await asyncio.wait_for(message_queue.get(), timeout=60)
        if message_type in ("start", "message"):
            print(f"Message received: {message}")
            yield message
        elif message_type == "function_result":
            print(f"Function result received: {message}")
        elif message_type == "function_error":
            print(f"Function error received: {message}")
        message_queue.task_done()
    except asyncio.TimeoutError:
        print(f"No messages received within the timeout period.")
        break

内容的提问来源于stack exchange，提问作者TheCyberAngel