需要访问旧版文档站？前往旧版文档官网首页。 (在新标签页打开)
声网文档中心

自定义大模型

更新时间不可用

了解如何在对话式 AI 引擎中自定义大模型，包括适用场景、配置步骤、实现要点和注意事项。

在 AI 对话互动场景下，你可能需要使用自定义的大语言模型 (Custom LLM) 实现更多个性化需求。

本文介绍如何使用自定义大模型接入声网对话式 AI 引擎。

技术原理

声网对话式 AI 引擎使用 OpenAI API 协议与 LLM 服务进行交互。接入自定义大模型的核心是提供一个与 OpenAI API 兼容的 HTTP 服务，这个服务需要能够接收和返回符合 OpenAI API 协议的请求和响应。

在此基础上，你可以实现更多的自定义功能，包括但不限于：

  • 使用检索增强生成（RAG）功能，让大模型从特定知识库中检索信息
  • 使用多模态能力，让大模型以文本和音频两种模态输出
  • 使用工具调用功能，让大模型调用外部工具
  • 使用 Function Calling 功能，让大模型以函数调用的形式返回结构化数据

前提条件

开始前，请确保你已经：

  • 参考实现对话式智能体实现了与 AI 智能体对话互动的基本逻辑
  • 拥有可访问的自定义大语言模型服务
  • 如需使用检索增强生成（RAG）功能，已准备好向量数据库或检索系统

实现方式

创建符合 OpenAI API 协议的服务

要成功接入声网对话式 AI 引擎，你的自定义大模型服务必须提供一个与 OpenAI Chat Completions API 兼容的接口。关键点如下：

  1. 接口路径：提供接收请求的端点 (Endpoint)，例如 https://your-custom-llm-service/chat/completions。
  2. 请求格式：接受与 OpenAI API 协议兼容的请求参数。
  3. 响应格式：返回与 OpenAI API 协议兼容、且符合 SSE 规范的流式响应。

下面的示例代码展示了如何实现一个符合 OpenAI API 协议的接口：

Python
class TextContent(BaseModel):
    type: str = "text"
    text: str

class ImageContent(BaseModel):
    type: str = "image"
    image_url: HttpUrl

class AudioContent(BaseModel):
    type: str = "input_audio"
    input_audio: Dict[str, str]

class ToolFunction(BaseModel):
    name: str
    description: Optional[str]
    parameters: Optional[Dict]
    strict: bool = False

class Tool(BaseModel):
    type: str = "function"
    function: ToolFunction


class ToolChoice(BaseModel):
    type: str = "function"
    function: Optional[Dict]


class ResponseFormat(BaseModel):
    type: str = "json_schema"
    json_schema: Optional[Dict[str, str]]


class SystemMessage(BaseModel):
    role: str = "system"
    content: Union[str, List[str]]


class UserMessage(BaseModel):
    role: str = "user"
    content: Union[str, List[Union[TextContent, ImageContent, AudioContent]]]


class AssistantMessage(BaseModel):
    role: str = "assistant"
    content: Union[str, List[TextContent]] = None
    audio: Optional[Dict[str, str]] = None
    tool_calls: Optional[List[Dict]] = None


class ToolMessage(BaseModel):
    role: str = "tool"
    content: Union[str, List[str]]
    tool_call_id: str

# 定义完整的请求格式
class ChatCompletionRequest(BaseModel):
    context: Optional[Dict] = None  # 上下文信息
    model: Optional[str] = None  # 使用的模型名称
    messages: List[Union[SystemMessage, UserMessage, AssistantMessage, ToolMessage]]  # 消息列表
    response_format: Optional[ResponseFormat] = None  # 响应格式
    modalities: List[str] = ["text"]  # 默认使用文本模态
    audio: Optional[Dict[str, str]] = None  # 助手的音频回复
    tools: Optional[List[Tool]] = None  # 工具列表
    tool_choice: Optional[Union[str, ToolChoice]] = "auto" # 工具选择
    parallel_tool_calls: bool = True  # 是否并行调用工具
    stream: bool = True  # 默认使用流式响应
    stream_options: Optional[Dict] = None  # 流式选项

@app.post("/chat/completions")
async def create_chat_completion(request: ChatCompletionRequest):
    try:
        logger.info(f"Received request: {request.model_dump_json()}")
        client = AsyncOpenAI(api_key=os.getenv("YOUR_LLM_API_KEY"))
        response = await client.chat.completions.create(
            model=request.model,
            messages=request.messages,  # 直接使用请求消息
            tool_choice=(
                request.tool_choice if request.tools and request.tool_choice else None
            ),
            tools=request.tools if request.tools else None,
            modalities=request.modalities,
            audio=request.audio,
            response_format=request.response_format,
            stream=request.stream,
            stream_options=request.stream_options,
        )
        if not request.stream:
            raise HTTPException(
                status_code=400, detail="chat completions require streaming"
            )

        async def generate():
            try:
                async for chunk in response:
                    logger.debug(f"Received chunk: {chunk}")
                    yield f"data: {json.dumps(chunk.to_dict())}\n\n"
                yield "data: [DONE]\n\n"
            except asyncio.CancelledError:
                logger.info("Request was cancelled")
                raise

        return StreamingResponse(generate(), media_type="text/event-stream")
    except asyncio.CancelledError:
        logger.info("Request was cancelled")
        raise HTTPException(status_code=499, detail="Request was cancelled")
    except Exception as e:
        traceback_str = "".join(traceback.format_tb(e.__traceback__))
        error_message = f"{str(e)}\n{traceback_str}"
        logger.error(error_message)
        raise HTTPException(status_code=500, detail=error_message)

配置声网对话式 AI 引擎

在调用 POST 创建对话式智能体时，将 LLM 配置指向你的自定义服务：

信息

如果访问你的自定义大模型服务需要校验身份，请在 api_key 字段中传入你的认证信息。

JSON
{
  "llm": {
    "url": "https://your-custom-llm-service/chat/completions",
    "api_key": "",
    "system_messages": [
      {
        "role": "system",
        "content": "You are a helpful assistant."
      }
    ]
  }
}

进阶功能

实现检索增强生成

如果你想提升智能体响应的准确性和相关性，可以利用检索增强生成（RAG）功能，让你的自定义大模型从特定知识库中检索信息，再将检索结果作为上下文提供给大模型生成回答。

以下示例代码模拟了从知识库中检索并返回内容的流程，并创建了 /rag/chat/completions 接口用于使用 RAG 检索结果调用 LLM 生成回答：

Python
async def perform_rag_retrieval(messages: Optional[Dict]) -> str:
    """
    使用 RAG 模型从知识库消息列表中检索相关内容

    Args:
        messages: 原始消息列表

    Returns:
        str: 检索到的文本内容
    """

    # TODO: 实现实际的 RAG 检索逻辑
    # 你可以根据具体需求从消息列表中选择第一个或最后一个消息作为查询，然后发送查询到 RAG 模型检索相关内容

    # 返回检索结果
    return "This is relevant content retrieved from the knowledge base."


def refact_messages(context: str, messages: Optional[Dict] = None) -> Optional[Dict]:
    """
    调整消息列表，将检索到的上下文添加到原始消息列表中

    Args:
        context: 检索到的上下文
        messages: 原始消息列表

    Returns:
        List: 调整后的消息列表
    """

    # TODO: 实现实际的消息调整逻辑
    # 这应该将检索到的上下文添加到原始消息列表中

    return messages

# 随机的等待消息
waiting_messages = [
    "Just a moment, I'm thinking...",
    "Let me think about that for a second...",
    "Good question, let me find out...",
]


@app.post("/rag/chat/completions")
async def create_rag_chat_completion(request: ChatCompletionRequest):
    try:
        logger.info(f"Received RAG request: {request.model_dump_json()}")
        if not request.stream:
            raise HTTPException(
                status_code=400, detail="chat completions require streaming"
            )

        async def generate():
            # First send a "please wait" prompt
            waiting_message = {
                "id": "waiting_msg",
                "choices": [
                    {
                        "index": 0,
                        "delta": {
                            "role": "assistant",
                            "content": random.choice(waiting_messages),
                        },
                        "finish_reason": None,
                    }
                ],
            }
            yield f"data: {json.dumps(waiting_message)}\n\n"

            # Perform RAG retrieval
            retrieved_context = await perform_rag_retrieval(request.messages)

            # Adjust messages
            refacted_messages = refact_messages(retrieved_context, request.messages)

            # Request LLM completion
            client = AsyncOpenAI(api_key=os.getenv("<YOUR_LLM_API_KEY>"))
            response = await client.chat.completions.create(
                model=request.model,
                messages=refacted_messages,
                tool_choice=(
                    request.tool_choice
                    if request.tools and request.tool_choice
                    else None
                ),
                tools=request.tools if request.tools else None,
                modalities=request.modalities,
                audio=request.audio,
                response_format=request.response_format,
                stream=True,  # Force streaming
                stream_options=request.stream_options,
            )

            try:
                async for chunk in response:
                    logger.debug(f"Received RAG chunk: {chunk}")
                    yield f"data: {json.dumps(chunk.to_dict())}\n\n"
                yield "data: [DONE]\n\n"
            except asyncio.CancelledError:
                logger.info("RAG stream was cancelled")
                raise

        return StreamingResponse(generate(), media_type="text/event-stream")

    except asyncio.CancelledError:
        logger.info("RAG request was cancelled")
        raise HTTPException(status_code=499, detail="Request was cancelled")
    except Exception as e:
        traceback_str = "".join(traceback.format_tb(e.__traceback__))
        error_message = f"{str(e)}\n{traceback_str}"
        logger.error(error_message)
        raise HTTPException(status_code=500, detail=error_message)

在调用 POST 创建对话式智能体时，只需将 LLM URL 指向你的 RAG 接口：

信息

如果访问你的自定义大模型服务需要校验身份，请在 api_key 字段中传入你的认证信息。

JSON
{
  "llm": {
    "url": "http://your-custom-llm-service/rag/chat/completions",
    "api_key": ""
    "system_messages": [
      {
        "role": "system",
        "content": "Please answer the user's question based on the following retrieved information: ..."
      }
    ]
  }
}

实现多模态能力

声网对话式 AI 引擎支持大模型以多模态形式（文本和音频）输出，你可以创建专用的多模态接口以实现更多个性化需求。

信息

如果你想了解更多关于使用音频模态输出的信息，你可以阅读使用音频模态输出。

以下示例代码展示了通过读取文本和音频文件，并发送给大模型生成音频回复的流程：

Python
async def read_text_file(file_path: str) -> str:
    """
    读取文本文件并返回内容

    Args:
        file_path: 文本文件的路径

    Returns:
        str: 文本文件的内容

    """
    async with aiofiles.open(file_path, "r") as file:
        content = await file.read()

    return content


async def read_pcm_file(
    file_path: str, sample_rate: int, duration_ms: int
) -> List[bytes]:
    """
    读取 PCM 文件并返回音频块列表

    Args:
        file_path: PCM 文件的路径
        sample_rate: 音频的采样率
        duration_ms: 每个音频块的时长，单位为毫秒

    Returns:
        List: 音频块列表

    """

    async with aiofiles.open(file_path, "rb") as file:
        content = await file.read()

    chunk_size = int(sample_rate * 2 * (duration_ms / 1000))
    return [content[i : i + chunk_size] for i in range(0, len(content), chunk_size)]


@app.post("/audio/chat/completions")
async def create_audio_chat_completion(request: ChatCompletionRequest):
    try:
        logger.info(f"Received audio request: {request.model_dump_json()}")

        if not request.stream:
            raise HTTPException(
                status_code=400, detail="chat completions require streaming"
            )

        # 示例用法，读取文本和音频文件
        # 请替换为你的实际逻辑

        text_file_path = "./file.txt"
        pcm_file_path = "./file.pcm"
        sample_rate = 16000  # 示例采样率
        duration_ms = 40  # 40ms 音频块

        text_content = await read_text_file(text_file_path)
        audio_chunks = await read_pcm_file(pcm_file_path, sample_rate, duration_ms)

        async def generate():
            try:
                # 发送文本内容
                audio_id = uuid.uuid4().hex
                text_message = {
                    "id": uuid.uuid4().hex,
                    "choices": [
                        {
                            "index": 0,
                            "delta": {
                                "audio": {
                                    "id": audio_id,
                                    "transcript": text_content,
                                },
                            },
                            "finish_reason": None,
                        }
                    ],
                }
                yield f"data: {json.dumps(text_message)}\n\n"

                # 发送音频块
                for chunk in audio_chunks:
                    audio_message = {
                        "id": uuid.uuid4().hex,
                        "choices": [
                            {
                                "index": 0,
                                "delta": {
                                    "audio": {
                                        "id": audio_id,
                                        "data": base64.b64encode(chunk).decode("utf-8"),
                                    },
                                },
                                "finish_reason": None,
                            }
                        ],
                    }
                    yield f"data: {json.dumps(audio_message)}\n\n"

                yield "data: [DONE]\n\n"

            except asyncio.CancelledError:
                logger.info("Audio stream was cancelled")
                raise

        return StreamingResponse(generate(), media_type="text/event-stream")

    except asyncio.CancelledError:
        logger.info("Audio request was cancelled")
        raise HTTPException(status_code=499, detail="Request was cancelled")
    except Exception as e:
        traceback_str = "".join(traceback.format_tb(e.__traceback__))
        error_message = f"{str(e)}\n{traceback_str}"
        logger.error(error_message)
        raise HTTPException(status_code=500, detail=error_message)

调用 POST 创建对话式智能体时，参考如下配置：

JSON
{
  "llm": {
    "url": "http://your-custom-llm-service/audio/chat/completions",
    "api_key": "your_api_key",
    "input_modalities": ["text"],
    "output_modalities": ["text", "audio"]
    "system_messages": [
      {
        "role": "system",
        "content": "You are a helpful assistant."
      }
    ]
  }
}

通过特殊首包响应传递参数

流式响应（SSE）场景下，LLM 的单次回复被切分为多个包 (Chunk) 进行传输。声网对话式 AI 引擎支持处理 object 为 chat.completion.custom_metadata 的特殊 LLM 首包响应，并根据特殊首包中的 metadata 字段实现以下高级功能：

为此，你需要对你的自定义大模型服务进行改造，使其输出符合对话式 AI 引擎要求的特殊首包响应。

注意

对话式 AI 引擎仅处理 object 为 chat.completion.custom_metadata 的首包响应，并忽略 choices 中的内容，后续响应视为正常响应，不适用于元数据配置。因此，请确保需要 TTS 播报的内容生成到第二及后续响应中。

以下时序图展示了用户、对话式 AI 引擎、自定义大模型之间的数据流转流程：

配置 LLM 响应是否可打断

对话式 AI 引擎支持让自定义大模型判断本次回复是否可被用户指令打断，从而确保智能体在播报法规、政策、产品定价等重要信息时，不被用户发言打断。

应用场景

  • 智能体播报重要法规、政策信息时，不允许被打断。
  • 智能体播报产品定价、合同条款等关键信息时，确保完整播报。

首包响应结构

你需要对你的自定义大模型服务进行改造，输出以下数据结构的首包响应：

JSON
{
    "id": "response-id",
    "object": "chat.completion.custom_metadata",
    "choices": [],
    "metadata": {
        "interruptable": false
    }
}

其中，interruptable 字段用于配置 TTS 播报本次 LLM 响应时是否能被用户发言打断：

  • interruptable 字段不存在，或 interruptable 字段值为空或值非法时，保持原打断模式不变。
  • interruptable 为 false，本次响应不允许被打断。
  • interruptable 为 true，本次响应允许被打断。

实时更新 TTS 参数

对话式 AI 引擎支持在对话中实时更新 TTS 参数，实现更沉浸的对话交互体验。

应用场景

  • 自定义大模型识别到用户要求智能体换一个声音/音色。
  • 自定义大模型识别到用户高兴的情绪时，实时提高 TTS 的音量、音调和语速等，使智能体的回应更符合用户的心情。

首包响应结构

你需要对你的自定义大模型服务进行改造，在大模型判断需要更新 TTS 参数时，让自定义大模型输出以下数据结构的首包响应：

JSON
{
    "id": "response-id",
    "object": "chat.completion.custom_metadata",
    "choices": [],
    "metadata": {
        "tts_params": {
			"params": {
				"voice_type": "female_1",
				"rate": 1.1
			}
		}
    }
}

其中，tts_params.params 字段用于配置 TTS 参数，你可以参考 TTS 供应商的官方文档获取你需要自定义的参数说明。

参考信息

示例项目

声网提供了开源的示例项目供你参考，你可以前往下载或查看其中的源代码。

Conversational-AI-Server-Sample

接口标准

自定义大模型服务必须兼容 OpenAI Chat Completions API 的接口标准：

  • 请求格式：包含模型、消息、工具调用配置等参数
  • 响应格式：包含模型生成的回复、元数据等信息
  • 流式响应：符合 SSE (Server-Sent Events) 规范

详细接口标准可参考：

咨询电话

400 632 6626

上海声网科技有限公司

Copyright © 2026 声网保留所有权利