使用 OpenAI SDK 在 Azure AI Foundry 上使用 DeepSeek 模型开发推理应用

了解如何将推理模型（如 Azure OpenAI 中的 DeepSeek）与适用于 Python 的 OpenAI SDK 配合使用。

本文介绍集成推理模型的几个最佳做法：

• 无密钥身份验证：使用托管标识或开发人员凭据，而不是 API 密钥。
• 异步作：使用异步功能来提高性能。
• 流式处理响应：向用户提供即时反馈。
• 推理分离：将推理步骤与最终输出分开。
• 资源管理：使用后清理资源。

DeepSeek 构成模块

浏览 DeepSeek 构建基块示例。 它演示如何使用 OpenAI 客户端库调用 DeepSeek-R1 模型并生成对用户消息的响应。

体系结构概述

下图显示了示例应用的简单体系结构：

聊天应用作为 Azure 容器应用运行。 应用程序使用 Microsoft Entra ID 的托管标识对 Azure OpenAI 进行身份验证，而不是使用 API 密钥。 应用使用 Azure OpenAI 生成对用户消息的响应。

应用依赖于以下服务和组件：

• 使用 OpenAI 客户端库包生成对用户消息的响应的 Python Quart 应用
• 基本的 HTML/JS 前端，通过 ReadableStream 使用 JSON Lines 从后端流式传输响应
• 用于预配 Azure 资源的 Bicep 文件，包括 Azure AI 服务、Azure 容器应用、Azure 容器注册表、Azure Log Analytics 和 RBAC 角色。

成本

为了降低成本，此示例对大多数资源使用基本或消耗定价层。 根据需要调整分层，并在完成后删除资源，以避免产生费用。

详细了解示例存储库 中的成本。

先决条件

开发容器包含本文所需的所有依赖项。 可以在 GitHub Codespaces（在浏览器中）或使用 Visual Studio Code 在本地运行它。

若要遵循本文，请确保满足以下先决条件：

开放开发环境

按照以下步骤设置具有所有必需依赖项的预配置开发环境。

部署和运行

示例存储库包含将聊天应用部署到 Azure 所需的所有代码和配置文件。 按照以下步骤将聊天应用部署到 Azure。

将聊天应用部署到 Azure

1. 运行以下 Azure Developer CLI 命令，以进行 Azure 资源预配和源代码部署：

   azd up
   

2. 使用下表回答提示：

   Prompt	答案
   环境名称	保持简短和小写。 添加名称或别名。 例如，chat-app。 它用作资源组名称的一部分。
   订阅	选择要在其中创建资源的订阅。
   位置（用于托管）	从列表中选择附近的位置。
   DeepSeek 模型的位置	从列表中选择附近的位置。 如果与第一个位置相同的位置可用，请选择该位置。

3. 等待应用部署。 部署通常需要 5 到 10 分钟。

使用聊天应用向大型语言模型提问

1. 部署后，终端会显示 URL。

2. 选择标记为 Deploying service web 在浏览器中打开聊天应用的 URL。

   

3. 在浏览器中，询问上传图像的问题，如“谁画了蒙娜丽莎？”

4. Azure OpenAI 通过模型推理提供答案，结果将显示在应用中。

探索示例代码

OpenAI 和 Azure OpenAI 服务都使用 常见的 Python 客户端库，但需要对 Azure OpenAI 终结点进行少量代码更改。 此示例使用 DeepSeek-R1 推理模型在简单的聊天应用中生成响应。

设置和身份验证

该文件 src\quartapp\chat.py 从设置和配置无密钥身份验证开始。

基础结构设置

该脚本使用 Quart（异步 Web 框架）来创建一个名为Blueprint的chat。 此 Blueprint 将定义应用的路由并管理其生命周期挂钩。

bp = Blueprint("chat", __name__, template_folder="templates", static_folder="static")

Blueprint 将定义 / 和 /chat/stream 路由，以及 @bp.before_app_serving 和 @bp.after_app_serving 生命周期挂钩。

使用无密钥身份验证进行初始化

以下代码片段处理身份验证。

身份验证策略适应环境：

• 在生产环境中：将 托管标识凭据 与 Azure 客户端 ID 配合使用，以避免存储敏感密钥。 此方法对于云原生应用是安全的且可缩放的。
• 在开发中：使用 Azure 开发人员 CLI 凭据和 Azure 租户 ID 来简化本地测试，方法是使用开发人员的 Azure CLI 登录会话。

@bp.before_app_serving
async def configure_openai():
    if os.getenv("RUNNING_IN_PRODUCTION"):
        client_id = os.environ["AZURE_CLIENT_ID"]
        bp.azure_credential = ManagedIdentityCredential(client_id=client_id)
    else:
        tenant_id = os.environ["AZURE_TENANT_ID"]
        bp.azure_credential = AzureDeveloperCliCredential(tenant_id=tenant_id)

此无密钥身份验证方法提供：

• 更好的安全性：代码或环境变量中未存储任何 API 密钥。
• 更易于管理：无需轮换密钥或管理机密。
• 平滑转换：相同的代码适用于开发和生产。

令牌提供程序设置

在以下代码片段中，令牌提供程序创建持有者令牌以对对 Azure OpenAI 服务的请求进行身份验证。 它使用配置的凭据自动生成并刷新这些令牌。

bp.openai_token_provider = get_bearer_token_provider(
    bp.azure_credential, "https://cognitiveservices.azure.com/.default"
)

Azure OpenAI 客户端配置

有两个可能的客户端， AzureOpenAI 以及 AsyncAzureOpenAI。 以下代码片段 AsyncAzureOpenAI 与异步 Quart 框架一起使用，以提高并发用户的性能：

bp.openai_client = AsyncAzureOpenAI(
    azure_endpoint=os.environ["AZURE_INFERENCE_ENDPOINT"],
    azure_ad_token_provider=openai_token_provider,
    api_version="2024-10-21",

• base_url：指向 Azure 托管的 DeepSeek 推理终结点
• api_key：使用根据令牌提供程序动态生成的 API 密钥。
• api-version：指定支持 DeepSeek 模型的 API 版本

模型部署名称配置

以下代码片段通过从环境配置获取部署名称来设置 DeepSeek 模型版本。 它将名称 bp.model_deployment_name 分配给变量，使其在整个应用中都可访问。 此方法允许在不更新代码的情况下更改模型部署。

bp.model_deployment_name = os.getenv("AZURE_DEEPSEEK_DEPLOYMENT")

生命周期管理

以下代码片段通过在应用程序关闭时关闭异步 Azure OpenAI 客户端来防止资源泄漏。 @bp.after_app_serving 挂钩可确保资源得到正确清理。

@bp.after_app_serving
async def shutdown_openai():
    await bp.openai_client.close()

聊天处理程序流式处理函数

该chat_handler()函数通过DeepSeek-R1路由管理用户与chat/stream模型的交互。 它实时将响应流式传输到客户端并处理它们。 该函数从 JSON 有效负载中提取消息。

流式处理实现

1. 该 response_stream 函数首先接受来自客户端的消息。

   • request_messages：该路由期望接收包含用户消息的 JSON 负载。

   @bp.post("/chat/stream")
   async def chat_handler():
      request_messages = (await request.get_json())["messages"]
   

2. 接下来，函数从 OpenAI API 流式传输响应。 它将系统消息（如“你是一个有用的助手”）与用户提供的消息组合在一起。

   @stream_with_context
   async def response_stream():
       all_messages = [
           {"role": "system", "content": "You are a helpful assistant."},
       ] + request_messages
   

3. 接着，该函数将创建流式处理聊天完成请求。

   该方法 chat.completions.create 将消息发送到 DeepSeek-R1 模型。 stream=True 参数将启用实时响应流式处理。

     chat_coroutine = bp.openai_client.chat.completions.create(
         model=bp.openai_model,
         messages=all_messages,
         stream=True,
     )
   

4. 以下代码片段处理 DeepSeek-R1 模型的流式响应，并处理错误。 它循环访问更新、检查有效选项，并将每个响应区块作为 JSON 行发送。 如果发生错误，它会记录错误，并在继续流时向客户端发送 JSON 错误消息。

   try:
       async for update in await chat_coroutine:
           if update.choices:
               yield update.choices[0].model_dump_json() + "\n"
       except Exception as e:
           current_app.logger.error(e)
           yield json.dumps({"error": str(e)}, ensure_ascii=False) + "\n"
   
   return Response(response_stream())
   

推理内容处理

虽然传统语言模型仅提供最终输出，但推理模型（例如 DeepSeek-R1）会展示他们的中间推理步骤。 这些步骤使它们适用于：

• 解决复杂问题
• 执行数学计算
• 处理多步骤逻辑推理
• 做出透明决策

在前端，事件处理程序 submit 在 index.html 中处理流式响应。 此方法允许你访问和显示模型推理步骤以及最终输出。

前端使用一个 ReadableStream 处理来自后端的流式响应。 它将推理内容与常规内容分开，在可展开的部分显示推理，以及主聊天区域中的最终答案。

分步细分

1. 启动流式处理请求

   此代码片段在 JavaScript 前端和 Python 后端之间创建连接，从而启用 DeepSeek-R1 的 Azure OpenAI 与无密钥身份验证的集成。

   const response = await fetch("/chat/stream", {
       method: "POST",
       headers: {"Content-Type": "application/json"},
       body: JSON.stringify({messages: messages})
   });
   

2. 初始化变量

   以下代码片段初始化变量以单独存储答案和想法。 这种分离有助于有效处理推理内容。

   let answer = "";
   let thoughts = "";    
   

3. 处理每个更新

   以下代码片段异步循环访问模型的响应区块。

   for await (const event of readNDJSONStream(response.body)) {
   

4. 检测和路由内容类型

   该脚本检查事件是否包含字段 delta 。 如果是这样，它将根据内容是推理内容还是常规内容来进行处理。

   if (!event.delta) {
        continue;
   }
   if (event.delta.reasoning_content) {
        thoughts += event.delta.reasoning_content;
        if (thoughts.trim().length > 0) {
            // Only show thoughts if they are more than just whitespace
            messageDiv.querySelector(".loading-bar").style.display = "none";
            messageDiv.querySelector(".thoughts").style.display = "block";
            messageDiv.querySelector(".thoughts-content").innerHTML = converter.makeHtml(thoughts);
        }
    } else if (event.delta.content) {
        messageDiv.querySelector(".loading-bar").style.display = "none";
        answer += event.delta.content;
        messageDiv.querySelector(".answer-content").innerHTML = converter.makeHtml(answer);
    }
   

   • 如果内容类型为reasoning_content，则内容将添加到thoughts中，并显示在.thoughts-content部分。
   • 如果内容类型为content，则内容将添加到answer中，并显示在.answer-content部分。
   • 内容开始流式处理后，.loading-bar 会被隐藏；如果有任何想法，将显示 .thoughts 部分。

5. 错误处理：

   错误记录在后端，并以 JSON 格式返回给客户端。

   except Exception as e:
       current_app.logger.error(e)
       yield json.dumps({"error": str(e)}, ensure_ascii=False) + "\n"
   

   此前端代码段将在聊天界面中显示错误消息。

   messageDiv.scrollIntoView();
   if (event.error) {
       messageDiv.innerHTML = "Error: " + event.error;
   }
   

清理 GitHub Codespaces

获取帮助

将问题记录到存储库的“问题”。

后续步骤