大模型API调用实战③-零基础保姆级教程

第 7 章流式输出 —— 打字机效果的真实现7.1 基本用法streamTrue只加一个参数返回值从响应对象变成可迭代的块chunk序列stream client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: 写一首关于程序员的四行小诗}], streamTrue, stream_options{include_usage: True}, # 让最后一个块带 usage 统计 ) collected [] for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: # 防御式判断见7.2 piece chunk.choices[0].delta.content print(piece, end, flushTrue) # 01教程的老朋友 collected.append(piece) if chunk.usage: # 最后一块携带账单 print(f\n[本次消耗 {chunk.usage.total_tokens} tokens]) full_answer .join(collected) # 攒出完整回答 —— 记历史时用它7.2 必知细节delta 不是 message流式块里是chunk.choices[0].delta.content本块新增的一小段不是.message.content。首块的 delta 可能只有 role 没 content末块带 usage 时choices为空列表——所以if chunk.choices and ...的防御写法不是啰嗦是必需。完整回答要自己拼流式模式没有现成的全文.join(collected)后再 append 进对话历史。底层是 SSEServer-Sent Events服务器在一条 HTTP 长连接上持续推送data: {...}文本行最后推data: [DONE]。SDK 帮你解析成了 chunk 对象。第 7 批你写 FastAPI 接口时就要亲手生产这种 SSE 流转发给前端——今天是消费者那天当生产者。推理模型的流式思考阶段的块走delta.reasoning_content回答阶段才走delta.content两个字段分开接。7.3 进阶预告流式 工具调用当模型在流式模式下决定调用工具时工具名和参数会被拆碎在多个 chunk 的delta.tool_calls里按 index 分组、arguments 逐段拼接需要累积重组。参考骨架现在看懂结构即可tool_calls {} # index - {id, name, arguments} for chunk in stream: if not chunk.choices: continue delta chunk.choices[0].delta if delta.content: print(delta.content, end, flushTrue) # 普通文本照常打印 for tc in (delta.tool_calls or []): # 工具调用碎片 slot tool_calls.setdefault(tc.index, {id: , name: , arguments: }) if tc.id: slot[id] tc.id if tc.function.name: slot[name] tc.function.name if tc.function.arguments:slot[arguments] tc.function.arguments # 循环结束后 tool_calls 里就是完整的调用申请正因为这套拼装麻烦本模块的 Agent 主循环先用非流式保证清晰流式作为综合项目的升级挑战。第 8 章异步与并发 —— AsyncOpenAI 批量处理8.1 场景100 条评论要做情感分析串行调用100 次 × 3 秒 5 分钟。异步并发import asyncio from openai import AsyncOpenAI aclient AsyncOpenAI(api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com) async def analyze(comment: str) - str: r await aclient.chat.completions.create( # await等待时让出CPU modeldeepseek-chat, messages[{role: system, content: 判断评论情感只输出正面/负面/中性}, {role: user, content: comment}], temperature0) return r.choices[0].message.content async def main(): comments [物流超快包装也好, 用了三天就坏了, 还行吧凑合, 客服态度太差了] * 5 results await asyncio.gather(*[analyze(c) for c in comments]) # 并发全上 for c, s in zip(comments[:4], results[:4]): print(f{s} ← {c}) asyncio.run(main())*[...]是 01 教程 10.4 的解包把列表拆成 gather 的多个参数。20 条并发总耗时 ≈ 最慢的一条。8.2 并发不能裸奔Semaphore 限流全量并发会撞平台限速429。生产标配——信号量控制同时在飞的请求数sem asyncio.Semaphore(5) # 最多同时5个请求在路上 async def analyze_limited(comment: str) - str: async with sem: # 满5个时第6个在这排队等空位 return await analyze(comment) # main 里改用: results await asyncio.gather(*[analyze_limited(c) for c in comments])async with sem的原理就是 上下文管理器进门占坑、出门还坑异常也保证归还。面试可复述版LLM 调用是 IO 密集任务用 asyncio AsyncOpenAI 并发可将批量任务耗时从求和降为取最大生产中必须用 Semaphore 限制并发度以配合平台 RPM/TPM 限额配合重试与退避构成完整的吞吐控制方案。第 9 章结构化输出实战 —— JSON Mode 与 Pydantic本章把最强的两层落地成代码。9.1 正式认识 PydanticPydantic 的BaseModel自带数据校验的 dataclass。定义一次结构校验、转换、报错全自动# pip install pydantic from pydantic import BaseModel, Field, ValidationError from typing import Literal class OrderIssue(BaseModel): order_id: str | None Field(None, description订单号如A123未提及则为null) issue_type: Literal[退款, 换货, 查询, 其它] # 只允许这四个值 urgent: bool Field(..., description用户是否表达紧急情绪) # ... 表示必填 # 校验成功字符串true还会被自动转成 bool True ok OrderIssue.model_validate_json({order_id:A123,issue_type:退款,urgent:true}) print(ok.issue_type, ok.urgent) # 退款 True # 校验失败错误信息精确到字段 —— 这正是回喂自修的好材料 try: OrderIssue.model_validate_json({issue_type:投诉,urgent:1}) except ValidationError as e: print(e.errors()[0][loc], e.errors()[0][msg]) # (issue_type,) 不在允许值中三件套记牢model_validate_jsonJSON字符串→对象并校验、model_dump对象→字典、model_json_schema。9.2 JSON Mode Pydantic 校验 失败回喂通用三件套response_format{type: json_object}让服务端保证输出语法合法的 JSON各主流平台通用但字段对不对仍要 Pydantic 把关不过关就把错误喂回去重试import json EXTRACT_SYSTEM 从用户消息中提取售后信息只输出 JSON 对象结构如下 {order_id: 订单号字符串或null, issue_type: 退款|换货|查询|其它, urgent: true或false} 注意必须输出 json 格式。 # JSON Mode 要求提示词中出现 json 字样并描述结构 def extract_order_issue(text: str, max_retries: int 2) - OrderIssue: messages [{role: system, content: EXTRACT_SYSTEM}, {role: user, content: text}] for attempt in range(max_retries 1): r client.chat.completions.create( modeldeepseek-chat, messagesmessages, response_format{type: json_object}, # 第①道锁语法合法 temperature0) raw r.choices[0].message.content try: return OrderIssue.model_validate_json(raw) # 第②道锁字段校验 except ValidationError as e: if attempt max_retries: raise messages.append({role: assistant, content: raw}) # 它的错误答卷 messages.append({role: user, # 批改意见回喂 content: f输出未通过校验{e.errors()}。请修正后重新只输出 JSON。}) result extract_order_issue(单号A123的鞋开胶了急死了必须退钱) print(result.model_dump()) # {order_id: A123, issue_type: 退款, urgent: True}更强的第③道锁部分平台支持 Schema 级强约束OpenAI 的client.chat.completions.parse(..., response_formatOrderIssue)可直接返回校验好的对象。国产平台支持度不一上面这套JSON Mode Pydantic 回喂在所有平台都能跑是更通用的职业方案平台支持 parse 时优先用它防御代码照留。