一次性输出全部结果

同步对话：先写个能跑的测试脚本

client = OpenAI(
    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="自己的",
)


def llm_chat(messages,model="qwen3.7-plus"):
    completion = client.chat.completions.create(
        # 模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        model=model,
        messages=messages
        # temperature=0.7,
        # max_completion_tokens = 100
    )
    return completion.choices[0].message.content

if __name__ == '__main__':
     messages= [{"content":"你是一个游戏专家","role":"system"},{"content":"最近比较多的射击类的游戏有哪些?","role":"user"}]
     res=llm_chat(messages)
     print(res)

• client 是整个对话的"发件人"，所有请求都从它发出去，全局只需要建一个。

• llm_chat() 是我们的核心工具函数：入参是对话数组 + 模型名，出参是模型吐出来的纯文本。把它单独抽出来，本地脚本、接口、定时任务都能复用。

• system 消息很关键，它决定模型的语气和能力边界；user 才是真正的问题。

• 注释掉的两行是常用旋钮：temperature 控随机性，max_completion_tokens 控长度，真上线建议都打开。

跑一下 python xxx.py，能打印出游戏列表就说明链路通了，可以往下走。

把同步问答封成异步接口

脚本跑通了，但总不能让前端每次都来跑你的 py 文件。下面用 FastAPI 把它变成一个 HTTP 接口。

client = AsyncOpenAI(
        # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        base_url="自己的",
    )

@llm_day01_router.post("/case1", summary="LLM-DAY01-CASE1")
async def case1_api(llmCase1Request: LLMCase1):
    completion = await  client.chat.completions.create(
        model="qwen3.7-plus",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": llmCase1Request.question}
        ],
        temperature=0.7,
    )

    ai_reply = completion.choices[0].message.content
    return {
        "code":1,
        "message":"请求成功",
        "data":{
            "ai_reply":ai_reply
                }
    }

• LLMCase1：用 Pydantic 定义的入参结构，前端 JSON 里只要带 question 字段就行，FastAPI 会自动校验类型。

• AsyncOpenAI vs OpenAI：接口场景必须用异步客户端。同步客户端在等模型返回的那几秒会霸占 worker，并发一上来就全堵了；异步的能把这段时间让给其他请求。

• case1_api 是真正处理请求的函数：await ...create(...) 发起调用，completion.choices[0].message.content 取正文，最后包成 {code, message, data} 这种前后端都舒服的格式返回。

• summary 是接口文档（Swagger）里显示的说明，方便别人看你的 API。

启动：uvicorn main:app --reload，然后 POST http://127.0.0.1:8000/case1，body 传 {"question": "你好"} 就能拿到回复。

流式输出

同步接口的问题是：用户点完发送，要等模型把整段话憋完才一次性返回，慢一点的模型能卡好几秒，体验很差。流式就是让字一个一个往外蹦。

打字机效果（测试脚本）

# 1. 准备工作：初始化客户端
client = OpenAI(
    # 建议通过环境变量配置API Key，避免硬编码。
    api_key=os.environ["DASHSCOPE_API_KEY"],
    # API Key与地域强绑定，请确保base_url与API Key的地域一致。
    base_url="1",
)

# 2. 发起流式请求
completion = client.chat.completions.create(
    model="qwen3.7-plus",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "请介绍一下自己"}
    ],
    stream=True,
    stream_options={"include_usage": True}
)

# 3. 处理流式响应
# 用列表暂存响应片段，最后 join 比逐次 += 字符串更高效
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}")

full_response = "".join(content_parts)
# print(f"\n--- 完整回复 ---\n{full_response}")

• stream=True 让服务端一边生成一边推数据，客户端边收边打，所以你会看到字是"流"出来的。

• delta.content 是这一小片增量内容，None 的时候代表这一帧没有文本（比如纯 usage 帧），所以用 or "" 兜底。

• stream_options={"include_usage": True} 是 optional 的，开了之后流尾会多一个 chunk.usage 帧，方便你统计花了多少 token——要算钱的话这个必须有。

• content_parts 收集全部片段，后面如果想把完整回答写库、做摘要，直接用 full_response。

把流式输出封成 SSE 接口

最后一步，把流式搬到接口上，用 SSE（Server-Sent Events）推给前端，前端就能做实时打字机了。

client = AsyncOpenAI(
        # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        base_url="htmode/v1",
    )


#流式输出
async def stream_chunk(user_question:str):
    # 2. 发起流式请求
    completion = await client.chat.completions.create(
        model="qwen3.7-plus",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": user_question}
        ],
        stream=True,
        stream_options={"include_usage": True}
    )
    async for chunk in completion:
        if chunk.choices:
            choice = chunk.choices[0]
            if choice.delta:
                delta = choice.delta
                if delta.content:
                    yield f"data: {delta.content}\n\n"
    yield "data: [DONE]\n\n"


@llm_day01_router.post("/case2", summary="LLM-DAY01-CASE2")
async def case2_api(llmCase2Request: LLMCase1):
    return StreamingResponse(
        content=stream_chunk(llmCase2Request.question),
        media_type="text/event-stream")

• stream_chunk() 是个 async 生成器（函数里有 yield）。它被 StreamingResponse 包起来后，FastAPI 会边调边把数据推给客户端，全程不占着内存攒完整结果。

• SSE 的报文格式有讲究：data: 内容\n\n，前缀 data: 加结尾两个换行，前端用 EventSource 或 fetch 读流时才能正确切片。少一个 \n 前端都可能收不到。

• yield "data: [DONE]\n\n" 是约定俗成的结束标记，前端拿到它就可以关掉加载动画、把输入框解锁。

• media_type="text/event-stream" 不能漏，它告诉 HTTP 这一路是事件流而不是普通 JSON。

• case2_api 本身很薄：拿到问题 → 交给生成器 → 用 StreamingResponse 包一下返回。真正的活都在 stream_chunk 里。