本文以一个「AI 求职助手」为例，完整演示如何调用通义千问大模型：从阿里云百炼开通与配置，到 Python 直连验证，再到 FastAPI 封装接口、接口测试、最后用 Vue3 + Element Plus 前端对接，实现「单轮对话」与「流式输出」两种模式。所有代码均来自真实可运行项目。

---

目录

• 一、整体架构与最终效果

• 二、阿里云百炼配置（DASHSCOPE_API_KEY）

• 三、Python 直接调用大模型

  • 3.1 单轮对话（case1.py）

  • 3.2 流式输出（case2.py）

• 四、FastAPI 封装 HTTP 接口

  • 4.1 请求体 Schema

  • 4.2 路由：单轮 + SSE 流式

  • 4.3 在 main.py 注册路由

• 五、接口测试（curl / Postman / Swagger）

• 六、前端对接（Vue3 + Element Plus）

  • 6.1 接口封装 llm.js

  • 6.2 Vite 代理配置

  • 6.3 豆包风格聊天面板组件

  • 6.4 页面与路由

• 七、常见问题与踩坑

• 八、总结

---

一、整体架构与最终效果

我们要做的是一个模仿「豆包」的 AI 对话页面，包含两种模式：

模式	传输方式	后端接口	体验
单轮对话	普通 POST，等待完整响应后一次性渲染	POST /llm-day01/case1	发问 → 等几秒 → 整段出现
流式输出	POST + SSE（text/event-stream）逐段推送	POST /llm-day01/case2	发问 → 文字逐字「打字机」式出现

整体链路如下：

 浏览器(Vue3)
    │  /api/llm-day01/case1   (axios, 单轮)
    │  /api/llm-day01/case2   (fetch+SSE, 流式)
    ▼
 Vite 代理 (/api → http://127.0.0.1:8000)
    ▼
 FastAPI 后端 (case1_api.py)
    ▼
 阿里云百炼 DashScope (通义千问 qwen-plus, OpenAI 兼容)

---

二、阿里云百炼配置（DASHSCOPE_API_KEY）

本项目通过阿里云百炼（原 DashScope）提供的 OpenAI 兼容接口 调用通义千问，底层库使用官方 openai SDK。

2.1 开通与获取 API Key

1. 登录 阿里云百炼控制台。

2. 开通「模型服务」并进入 API-KEY 管理，点击「创建 API-KEY」。

3. 复制生成的 Key（格式形如 sk-xxxxxxxxxxxxxxxx）。

2.2 配置环境变量（推荐）

强烈建议用环境变量，不要硬编码 Key 到代码里。

 # Linux / macOS
 export DASHSCOPE_API_KEY="sk-你的key"
 ​
 # Windows (PowerShell)
 $env:DASHSCOPE_API_KEY="sk-你的key"
 ​
 # 或写入 .env 文件（不要提交到 git）
 echo DASHSCOPE_API_KEY=sk-你的key > .env

代码中直接读取：

 import os
 api_key = os.getenv("DASHSCOPE_API_KEY")

2.3 base_url 与模型

OpenAI 兼容模式下，百炼的 base_url 为：

 https://dashscope.aliyuncs.com/compatible-mode/v1

注：本文项目代码中使用的是专属 MaaS endpoint（https://ws-xxxxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1），功能等价，可按你自己的专属地址替换。

常用模型（通过 model 参数指定）：

• qwen-plus：通义千问 plus，性价比高（本项目使用）

• qwen-max：效果更强

• qwen-turbo：速度最快、最便宜

---

三、Python 直接调用大模型

先不碰 Web 框架，用纯 Python 验证大模型调用是否跑通。项目把这两段放在 llm/ 目录下。

3.1 单轮对话（case1.py）

 # llm/case1.py
 import os
 from openai import OpenAI
 ​
 client = OpenAI(
     api_key=os.getenv("DASHSCOPE_API_KEY"),
     base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
 )
 ​
 completion = client.chat.completions.create(
     model="qwen-plus",
     messages=[
         {"role": "system", "content": "You are a helpful assistant."},
         {"role": "user", "content": "人为什么要睡觉?"},
     ],
     temperature=0.75,
     max_completion_tokens=100,
 )
 ​
 print(completion.choices[0].message.content)

运行：

 pip install openai
 python llm/case1.py

3.2 流式输出（case2.py）

关键参数是 stream=True，此时返回的是一个可迭代的生成器，每收到一个 chunk 就 yield 一次。

 # llm/case2.py
 import os
 from openai import OpenAI
 ​
 client = OpenAI(
     api_key=os.getenv("DASHSCOPE_API_KEY"),
     base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
 )
 ​
 completion = client.chat.completions.create(
     model="qwen-plus",
     messages=[
         {"role": "system", "content": "You are a helpful assistant."},
         {"role": "user", "content": "请介绍一下自己"},
     ],
     stream=True,
     stream_options={"include_usage": True},  # 返回 token 用量
 )
 ​
 content_parts = []
 print("AI: ", end="", flush=True)
 ​
 for chunk in completion:
     if chunk.choices:
         content = chunk.choices[0].delta.content or ""
         print(content, end="", flush=True)
         content_parts.append(content)
     elif chunk.usage:
         print("\n--- 请求用量 ---")
         print(f"输入 Tokens: {chunk.usage.prompt_tokens}")
         print(f"输出 Tokens: {chunk.usage.completion_tokens}")
         print(f"总计 Tokens: {chunk.usage.total_tokens}")
 ​
 print(f"\n--- 完整回复 ---\n{''.join(content_parts)}")

流式模式下，内容通过 chunk.choices[0].delta.content 逐段获取；用它来实现「打字机」效果是标准做法。

---

四、FastAPI 封装 HTTP 接口

验证通了之后，把调用逻辑封装成 Web 接口，供前端调用。

4.1 请求体 Schema

使用 Pydantic 定义请求体，只接收一个 question 字段：

 # app/schemas/llm_case1.py
 from pydantic import BaseModel, Field
 ​
 ​
 class LLMCase1(BaseModel):
     question: str = Field(..., title="问题", description="用户问题")

4.2 路由：单轮 + SSE 流式

# app/llm/case1_api.py
import os

from fastapi import APIRouter
from openai import OpenAI
from starlette.responses import StreamingResponse

from app.schemas.llm_case1 import LLMCase1

BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"


@router = APIRouter(prefix="/llm-day01", tags=["LLM-DAY01"])


# ===== 单轮对话：等待完整回复后返回 JSON =====
@router.post("/case1", summary="单轮对话")
async def case1_api(body: LLMCase1):
    client = OpenAI(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        base_url=BASE_URL,
    )
    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=[
            {"role": "system", "content": "你是一个智能助手"},
            {"role": "user", "content": body.question},
        ],
        temperature=0.75,
    )
    ai_reply = completion.choices[0].message.content
    return {
        "code": 1,
        "message": "请求成功",
        "data": {"ai_reply": ai_reply},
    }


# ===== 流式对话：SSE 逐段推送 =====
def stream_chunk(user_question: str):
    client = OpenAI(
        api_key=os.environ["DASHSCOPE_API_KEY"],
        base_url=BASE_URL,
    )
    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": user_question},
        ],
        stream=True,
        stream_options={"include_usage": True},
    )
    for chunk in completion:
        if chunk.choices:
            delta = chunk.choices[0].delta
            if delta.content:
                yield f"data: {delta.content}\n\n"   # SSE 格式：data: 内容 + 两个换行

    yield "data: [done]\n\n"   # 自定义结束标志


@router.post("/case2", summary="流式对话(SSE)")
async def case2_api(body: LLMCase1):
    return StreamingResponse(
        content=stream_chunk(body.question),
        media_type="text/event-stream",
    )

SSE 格式要点：每个事件以 data: 内容\n\n（data: + 内容 + 两个换行）分隔，结束用 data: [done]\n\n。前端据此切分。

4.3 在 main.py 注册路由

# main.py（节选）
from app.llm.case1_api import llm_day01_router

app.include_router(llm_day01_router)

# 别忘了跨域，方便前端本地联调
from starlette.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    allow_credentials=True,
)

启动后端：

pip install fastapi uvicorn
uvicorn main:app --host 127.0.0.1 --port 8000 --reload

---

五、接口测试（curl / Postman / Swagger）

5.1 curl 测试单轮

curl -X POST http://127.0.0.1:8000/llm-day01/case1 \
  -H "Content-Type: application/json" \
  -d '{"question":"你好"}'

返回：

{
  "code": 1,
  "message": "请求成功",
  "data": {
    "ai_reply": "你好！很高兴见到你～ 有什么可以帮你的吗？"
  }
}

5.2 curl 测试流式

curl -N -X POST http://127.0.0.1:8000/llm-day01/case2 \
  -H "Content-Type: application/json" \
  -d '{"question":"介绍一下你自己"}'

你会看到内容被逐行推出来，最后一行是 data: [done]。

5.3 Swagger 在线文档

FastAPI 自带交互式文档，浏览器打开即可填参数直接测试：

http://127.0.0.1:8000/docs

如果接口文档里测试通过、但前端报错，问题几乎一定在前端（超时 / 代理 / SSE 解析），见第七节。

---

六、前端对接（Vue3 + Element Plus）

前端是标准的 Vue3 + Vite + Element Plus 项目。核心难点是流式接口是 POST，不能用浏览器的原生 EventSource（它只支持 GET），因此要手写 fetch + ReadableStream 解析 SSE。

6.1 接口封装 llm.js

// src/api/llm.js
import request from '@/utils/request'

// 单轮对话
export function askOnce(question) {
  return request({
    url: '/llm-day01/case1',
    method: 'post',
    data: { question },
    timeout: 120000, // 大模型单轮完整生成可能较慢，务必放宽超时
  })
}

// 流式对话（POST + SSE，不能用 EventSource）
export async function askStream(question, { onChunk, onDone, onError } = {}) {
  try {
    const token = localStorage.getItem('candidateToken')
    const resp = await fetch('/api/llm-day01/case2', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        ...(token ? { Authorization: `Bearer ${token}` } : {}),
      },
      body: JSON.stringify({ question }),
    })

    if (!resp.ok || !resp.body) {
      const text = await resp.text().catch(() => '')
      throw new Error(text || `请求失败（状态码 ${resp.status}）`)
    }

    const reader = resp.body.getReader()
    const decoder = new TextDecoder('utf-8')
    let buffer = ''

    while (true) {
      const { done, value } = await reader.read()
      if (done) break
      buffer += decoder.decode(value, { stream: true })

      // SSE 以空行(\n\n)分隔事件
      let sep
      while ((sep = buffer.indexOf('\n\n')) !== -1) {
        const event = buffer.slice(0, sep)
        buffer = buffer.slice(sep + 2)

        const dataLine = event
          .split('\n')
          .find((line) => line.startsWith('data:'))
        if (!dataLine) continue

        const data = dataLine.slice(5).trim()
        if (data === '[DONE]' || data === '[done]') {
          onDone && onDone()
          return
        }
        onChunk && onChunk(data)
      }
    }
    onDone && onDone()
  } catch (err) {
    console.error('[askStream] 流式请求异常:', err)
    onError && onError(err)
  }
}

6.2 Vite 代理配置

把 /api 代理到后端 8000，并去掉 /api 前缀：

// vite.config.js
export default defineConfig({
  server: {
    port: 3003,
    proxy: {
      '/api': {
        target: 'http://127.0.0.1:8000',
        changeOrigin: true,
        rewrite: (path) => path.replace(/^\/api/, ''),
      },
    },
  },
})

这样前端访问 /api/llm-day01/case1 实际打到 http://127.0.0.1:8000/llm-day01/case1，跨域问题交给代理解决。

6.3 豆包风格聊天面板组件

核心是一个可复用组件 AiChatPanel.vue，通过 mode 区分单轮 / 流式：

<!-- src/components/ai/AiChatPanel.vue（核心逻辑节选） -->
<script setup>
import { ref, reactive } from 'vue'
import { askOnce, askStream } from '@/api/llm'

const props = defineProps({ mode: { type: String, default: 'single' } })

const messages = ref([])
const inputText = ref('')
const loading = ref(false)

const sendSingle = async (question) => {
  loading.value = true
  const aiMsg = reactive({ role: 'ai', content: '', streaming: true })
  messages.value.push(aiMsg)
  try {
    const res = await askOnce(question)
    aiMsg.content = res?.data?.ai_reply || '（暂无回复）'
  } catch (e) {
    console.error('[sendSingle] 单轮请求失败:', e)
    aiMsg.content = '请求失败，请稍后重试。'
  } finally {
    aiMsg.streaming = false
    loading.value = false
  }
}

const sendStream = (question) => {
  loading.value = true
  const aiMsg = reactive({ role: 'ai', content: '', streaming: true })
  messages.value.push(aiMsg)
  askStream(question, {
    onChunk: (chunk) => { aiMsg.content += chunk },      // 逐字追加
    onDone: () => { aiMsg.streaming = false; loading.value = false },
    onError: (err) => {
      aiMsg.streaming = false
      aiMsg.content = aiMsg.content || '请求失败，请稍后重试。'
      loading.value = false
    },
  })
}
</script>

6.4 页面与路由

两个页面分别复用面板，再注册路由：

// router/index.js
import AiSingleChat from '@/pages/ai/AiSingleChat.vue'
import AiStreamChat from '@/pages/ai/AiStreamChat.vue'

// 在 /candidate 子路由 children 中增加：
{ path: 'ai-single', name: 'AiSingleChat', component: AiSingleChat },
{ path: 'ai-stream', name: 'AiStreamChat', component: AiStreamChat },

<!-- src/pages/ai/AiSingleChat.vue -->
<template>
  <AiChatPanel mode="single" />
</template>
<script setup>
import AiChatPanel from '@/components/ai/AiChatPanel.vue'
</script>

<!-- src/pages/ai/AiStreamChat.vue -->
<template>
  <AiChatPanel mode="stream" />
</template>
<script setup>
import AiChatPanel from '@/components/ai/AiChatPanel.vue'
</script>

启动前端：

npm install
npm run dev   # 默认 http://127.0.0.1:3003

---

七、常见问题与踩坑

❌ 坑 1：单轮对话「请求超时」

axios 默认 timeout: 15000（15 秒）。大模型在并发高或首字延迟大时，单轮完整响应很容易超过 15 秒，触发超时。

解决：在 askOnce 中单独设置 timeout: 120000（2 分钟），见 6.1。

流式模式不会超时，因为首字节很快返回，连接一直「活跃」。这也解释了为什么「流式正常、单轮超时」。

❌ 坑 2：流式接口是 POST，EventSource 用不了

new EventSource(url) 只能发 GET。我们的流式接口是 POST /llm-day01/case2，必须带 body。

解决：用 fetch + response.body.getReader() 手动读取流，按 \n\n 切分 SSE 事件，见 6.1 的 askStream。

❌ 坑 3：SSE 结束标志要自己约定

标准 SSE 用 data: [DONE] 结束。本项目后端约定 data: [done]（小写）。前端解析时两者都兼容即可。

❌ 坑 4：跨域 / 代理

本地开发若前端直接请求 http://127.0.0.1:8000 会触发 CORS。统一走 Vite 的 /api 代理（见 6.2）即可，无需在前端写完整域名。

❌ 坑 5：本地联调 405

用浏览器或 curl 以 GET 方式访问 /llm-day01/case1 会返回 405 Method Not Allowed——这是正常的，因为接口只接受 POST。测试请用 curl -X POST 或 Swagger。

---

八、总结

本文从配置到落地，打通了「阿里云百炼 → FastAPI → Vue3」的完整链路：

1. 配置：在百炼控制台拿到 DASHSCOPE_API_KEY，用环境变量注入，base_url 走 OpenAI 兼容模式。

2. Python 直连：stream=True 实现流式，delta.content 取片段。

3. FastAPI 封装：单轮返回 JSON，流式用 StreamingResponse + SSE（data: 片段 / data: [done]）。

4. 测试：curl、Postman、Swagger /docs 三件套验证接口。

5. 前端： axios 单轮（注意超时），fetch + ReadableStream 解析 POST 流式，Vite 代理解决跨域。

6. 踩坑：超时、EventSource 不支持 POST、SSE 结束标志、代理与 405。

照着这套，你也能快速给自己的系统加上一个「会打字」的 AI 助手。

---

环境依赖

 # 后端
 pip install fastapi uvicorn openai
 ​
 # 前端
 npm install axios element-plus

完整项目结构（节选）

 fastApiProject4/
 ├── llm/
 │   ├── case1.py          # Python 单轮直连示例
 │   └── case2.py          # Python 流式直连示例
 ├── app/
 │   ├── llm/case1_api.py  # FastAPI 单轮 + SSE 流式接口
 │   ├── schemas/llm_case1.py
 │   └── main.py           # 注册路由 + CORS
 └── main.py
 ​
 new_boss_vue-main/boss-candidate-ui/
 ├── src/api/llm.js          # 接口封装（单轮 + 流式）
 ├── src/components/ai/AiChatPanel.vue
 ├── src/pages/ai/AiSingleChat.vue
 ├── src/pages/ai/AiStreamChat.vue
 ├── src/router/index.js
 └── vite.config.js          # /api 代理到 8000