长谷川

菜单

Q
发布于 2025-07-21 / 6 阅读
0
0

使用 OpenAI 调用 AI 模型：流式与非流式指南

#后端笔记

准备工作

在开始之前，请确保你已完成以下准备：

  1. 安装 Deno: 如果你尚未安装 Deno，请访问 Deno 官方网站 获取安装指南。

  2. 获取 AI API Key: 你需要从 AI 模型提供商（例如 Kimi AI / Moonshot AI）的官网获取你的 API Key。请妥善保管此 Key。 *

  3. 创建项目文件:

    • 在你的项目根目录下创建一个 TypeScript 文件，例如 ai_client.ts。

    • 在同一目录下创建一个 .env 文件，用于安全存储你的 API Key。

.env 文件内容

在 .env 文件中，添加你的 AI API Key：

MOONSHOT_API_KEY=sk-YOUR_API_KEY_HERE

请将 sk-YOUR_API_KEY_HERE 替换为你的实际 API Key。

核心依赖

我们将使用 Deno 的 dotenv 模块来加载环境变量，以及 openai 模块来与 AI 模型 API 交互。

import { load } from "https://deno.land/std@0.224.0/dotenv/mod.ts";
import OpenAI from 'https://deno.land/x/openai@v4.52.0/mod.ts';

1. 非流式调用 AI 模型

非流式（Non-Streaming）是最直接的调用方式。你的程序会发送请求，然后等待 AI 模型生成完整的回答后，一次性接收并显示出来。

callAI 函数（非流式版本）

以下是实现非流式调用的 callAI 函数：

import { load } from "https://deno.land/std@0.224.0/dotenv/mod.ts";
import OpenAI from 'https://deno.land/x/openai@v4.52.0/mod.ts';

/**
 * 非流式调用 AI 模型获取回答。
 *
 * @param question 用户的问题。
 * @param system 系统指令，默认为“你是一个乐于助人的AI大模型”。
 * @returns AI 的完整回答字符串，或错误信息。
 */
async function callAI_NonStreaming(question: string, system: string = "你是一个乐于助人的AI大模型"): Promise<string | undefined> {
  // 加载 .env 文件中的环境变量
  const env = await load();
  const MOONSHOT_API_KEY = env.MOONSHOT_API_KEY;

  if (!MOONSHOT_API_KEY) {
    console.error('❌ 错误：未找到 MOONSHOT_API_KEY 环境变量。请确保在 .env 文件中设置。');
    Deno.exit(1);
  }

  // 配置 OpenAI 客户端指向 AI API
  const openai = new OpenAI({
    apiKey: MOONSHOT_API_KEY,
    baseURL: "https://api.moonshot.cn/v1", // Kimi AI / Moonshot AI 的 API 基础 URL
  });

  try {
    // 发送请求，stream: false 表示非流式
    const chatCompletion = await openai.chat.completions.create({
      model: "moonshot-v1-8k", // AI 的聊天模型名称，可根据需求更改
      messages: [
        { role: "system", content: system },
        { role: "user", content: question }
      ],
      stream: false, // 非流式模式
    });

    // 提取 AI 的回答内容
    const aiContent = chatCompletion.choices[0]?.message?.content;

    // 返回 AI 的回答
    return aiContent;

  } catch (error) {
    console.error("调用 AI API 时发生错误：", error);
    if (error instanceof OpenAI.APIError) {
      console.error(`状态码: ${error.status}`);
      console.error(`错误类型: ${error.type}`);
      console.error(`消息: ${error.message}`);
      console.error(`代码: ${error.code}`);
    }
    return "抱歉，AI 回答失败。"; // 失败时返回一个错误消息
  }
}

/**
 * 主程序入口点，演示非流式 AI 问答。
 */
async function mainNonStreaming() {
  console.log("--- 非流式 AI 问答程序 ---");
  console.log("欢迎使用 AI 问答程序！输入 '退出' 或 'exit' 结束。");

  while (true) {
    const userQuestion = prompt("你有什么问题？🤔 ");

    if (userQuestion === null || userQuestion.toLowerCase() === "退出" || userQuestion.toLowerCase() === "exit") {
      console.log("程序结束。👋");
      break;
    }

    console.log("AI 思考中... 请稍候...");
    // 调用非流式函数获取完整回答
    const aiAnswer = await callAI_NonStreaming(userQuestion);

    console.log("AI 回答: ✨");
    console.log(aiAnswer); // 打印完整回答
    console.log("------------------------------------");
  }
}

async function main() {
    console.log("欢迎使用 AI 问答程序！输入 '退出' 或 'exit' 结束。");
  
    while (true) {
      // 1. 在终端中提示用户输入问题
      const userQuestion = prompt("你有什么问题？🤔 ");
  
      // 2. 检查用户是否想要退出
      if (userQuestion === null || userQuestion.toLowerCase() === "退出" || userQuestion.toLowerCase() === "exit") {
        console.log("程序结束。👋");
        break; // 退出循环
      }
  
      // 3. 调用 AI 函数获取回答
      console.log("AI 思考中... 请稍候...");
      const aiAnswer = await callAI_NonStreaming(userQuestion);
  
      // 4. 在终端中打印 AI 的回答
      console.log("AI 回答: ✨");
      console.log(aiAnswer);
      console.log("------------------------------------");
    }
  }
  
  // 运行主函数
  main();

运行非流式程序

将上述代码保存为 ai_client.ts。在终端中运行：

deno run --allow-env --allow-read --allow-prompt ai_client.ts

或者在 deno.json 中配置

"tasks": {
    "dev": "deno run --allow-env --allow-read --allow-write --allow-net --watch main.ts"
  },

然后deno run dev

运行后，程序会等待你输入问题。输入问题后，你会看到“AI 思考中...”，然后等待一段时间后，AI 的完整回答会一次性显示出来。

2. 流式调用 AI 模型

流式（Streaming）调用模式允许你实时地接收 AI 模型生成的内容，就像打字一样，一个字一个字地显示在终端上。这大大提升了用户体验，尤其是在模型生成较长回答时。

为了实现流式输出，callAI 函数需要变成一个异步生成器函数（async function*），它会使用 yield 关键字分段生成内容。

callAI 函数（流式版本）

以下是实现流式调用的 callAI 函数：

import { load } from "https://deno.land/std@0.224.0/dotenv/mod.ts";
import OpenAI from 'https://deno.land/x/openai@v4.52.0/mod.ts';

// 将 function 改为 async function*
async function* callAI_Streaming(question: string, system: string = "你是一个乐于助人的AI大模型"): AsyncGenerator<string, void, void> { // <-- 注意这里的 *
  const env = await load();
  const MOONSHOT_API_KEY = env.MOONSHOT_API_KEY;

  if (!MOONSHOT_API_KEY) {
    console.error('❌ 错误：未找到 MOONSHOT_API_KEY 环境变量。请确保在 .env 文件中设置。');
    Deno.exit(1);
  }

  const openai = new OpenAI({
    apiKey: MOONSHOT_API_KEY,
    baseURL: "https://api.moonshot.cn/v1",
  });

  try {
    const stream = await openai.chat.completions.create({
      model: "moonshot-v1-8k",
      messages: [
        { role: "system", content: system },
        { role: "user", content: question }
      ],
      stream: true, // <-- 关键改变：设置为 true
    });

    // 异步迭代流式响应
    for await (const chunk of stream) {
      // 从每个数据块中提取内容，并生成 (yield) 它
      // 注意：OpenAI 库对于流式响应的 chunk.choices[0]?.delta?.content 属性是关键
      const content = chunk.choices[0]?.delta?.content || "";
      yield content; // <-- 生成每次接收到的内容片段
    }

  } catch (error) {
    console.error("调用 AI API 时发生错误：", error);
    if (error instanceof OpenAI.APIError) {
      console.error(`状态码: ${error.status}`);
      console.error(`错误类型: ${error.type}`);
      console.error(`消息: ${error.message}`);
      console.error(`代码: ${error.code}`);
    }
    // 发生错误时，可以生成一个错误提示，或者抛出错误
    yield "抱歉，AI 回答失败。";
  }
}

// ... (callAI_Streaming函数定义，如上所示) ...

async function main() {
    console.log("欢迎使用 AI 问答程序！输入 '退出' 或 'exit' 结束。");

    while (true) {
      const userQuestion = prompt("你有什么问题？🤔 ");

      if (userQuestion === null || userQuestion.toLowerCase() === "退出" || userQuestion.toLowerCase() === "exit") {
        console.log("程序结束。👋");
        break;
      }

      console.log("AI 思考中... 请稍候...");
      console.log("AI 回答: ✨");

      // 关键改变：使用 for await...of 循环处理流
      const aiResponseStream = callAI_Streaming(userQuestion); // callAI 返回一个异步迭代器

      try {
        for await (const chunk of aiResponseStream) {
          // 每次收到一小块内容就立即打印
          Deno.stdout.write(new TextEncoder().encode(chunk)); // Deno.stdout.write 用于不换行打印
        }
        console.log("\n------------------------------------"); // 回答结束后换行和分隔符
      } catch (error) {
        console.error("\n处理流时发生错误:", error);
        console.log("------------------------------------");
      }
    }
}

// 运行主函数
main();

运行流式程序

将上述代码保存为 ai_client.ts。在终端中运行：

deno run --allow-env --allow-read --allow-prompt ai_client.ts

或者在 deno.json 中配置

"tasks": {
    "dev": "deno run --allow-env --allow-read --allow-write --allow-net --watch main.ts"
  },

然后deno run dev

运行后，程序会等待你输入问题。输入问题后，你会看到 AI 的回答像打字一样，一个字一个字地实时显示出来，直到回答结束。

流式与非流式对比

特性

非流式

流式

用户体验

等待完整回答，然后一次性显示。可能感觉延迟。

实时显示，逐步呈现。用户体验更流畅，减少等待感。

实现复杂度

相对简单，直接返回字符串。

涉及异步生成器（async function*）和异步迭代（for await...of），稍复杂。

资源利用

模型生成完毕后才释放连接。

数据边生成边传输，可能更早释放部分资源。

适用场景

简单、短小、对实时性要求不高的回答。

长文本生成、聊天机器人、需要即时反馈的应用。

何时选择？

  • 如果需要快速获取一个简短的、完整的回答，并且对实时性要求不高，非流式可能更简单。

  • 聊天机器人等需要生成较长的内容，或者希望提供更流畅的用户体验，流式是更好的选择。

感觉一般都用流式

Object.defineProp...
正则表达式：文本处理的瑞士军刀

评论