🐍 Tutorial 03 Real API
📄 文件:
⬇ 下载 .py 文件tutorial_03_real_api.py"""
============================================================
真实 LLM API 实战教程（OpenAI 兼容：DeepSeek / Qwen / Moonshot…）
============================================================
目标：把 Day 4/6 的 mock 换成真实 API，打通"练习 → 实战"的最后一公里
运行方式：python tutorial_03_real_api.py
环境要求：Python 3.11+，pip install aiohttp
API Key（可选：没有 Key 也能学第 1-3 步，配好后运行可看全部 8 步）：
申请地址：https://platform.deepseek.com （有免费额度）
配置方式——设置环境变量（推荐，永远不要把 Key 写进代码！）：
Windows PowerShell: $env:DEEPSEEK_API_KEY = "sk-..."
Windows CMD: set DEEPSEEK_API_KEY=sk-...
macOS / Linux: export DEEPSEEK_API_KEY=sk-...
也兼容 OPENAI_API_KEY / LLM_API_KEY 环境变量名。
⚠️ 安全三条铁律：
1. Key 只放环境变量或 .env（.env 必须进 .gitignore）
2. 报错信息里永远不要打印完整 Key（本教程会演示脱敏）
3. 所有真实调用都要设 max_tokens 和超时，防止费用失控
============================================================
"""
import asyncio
import json
import os
import time
try:
import aiohttp
except ImportError:
aiohttp = None
BASE_URL = "https://api.deepseek.com" # Qwen/Moonshot 等只需换这里的域名和模型名
MODEL = "deepseek-chat"
# 全局成本统计（第 8 步汇总）
usage_total = {"prompt_tokens": 0, "completion_tokens": 0, "calls": 0}
def 分隔线(标题):
print("\n" + "=" * 60)
print(标题)
print("=" * 60)
def 找key() -> str:
"""按优先级找 API Key——第 1 步的核心：Key 只从环境变量来"""
for 名称 in ("DEEPSEEK_API_KEY", "OPENAI_API_KEY", "LLM_API_KEY"):
值 = os.environ.get(名称, "").strip()
if 值:
return 值
return ""
def 脱敏(key: str) -> str:
"""打印时永远脱敏：sk-ab****yz"""
if len(key) <= 8:
return "****"
return f"{key[:5]}****{key[-4:]}"
# ==================== 请求/解析（离线部分也能学） ====================
def 构造请求体(messages, tools=None, stream=False, max_tokens=512):
"""第 2 步：OpenAI 兼容的请求体结构——每个字段都要认识"""
body = {
"model": MODEL,
"messages": messages, # 对话历史：system/user/assistant/tool
"max_tokens": max_tokens, # 必设！防止费用失控
"temperature": 0.7,
"stream": stream,
}
if tools:
body["tools"] = tools # Function Calling：告诉 LLM 有哪些工具
return body
def 解析响应(resp_json):
"""第 3 步：从响应 JSON 里取出回答和 token 用量"""
message = resp_json["choices"][0]["message"]
usage = resp_json.get("usage", {})
return message, usage
async def call_llm(session, messages, tools=None, stream=False, max_tokens=512, 轮次说明=""):
"""第 3+4 步：带完整错误处理的 LLM 调用（超时/状态码/重试/退避）"""
body = 构造请求体(messages, tools=tools, stream=stream, max_tokens=max_tokens)
headers = {"Authorization": f"Bearer {找key()}", "Content-Type": "application/json"}
最大重试 = 2
last_err = None
for attempt in range(最大重试 + 1):
try:
async with asyncio.timeout(20): # 每次请求都有超时
async with session.post(
f"{BASE_URL}/chat/completions", json=body, headers=headers
) as resp:
# HTTP 状态码分诊：不同错误不同处理
if resp.status == 401:
raise RuntimeError("401：API Key 无效或过期，请检查环境变量")
if resp.status == 429:
raise RuntimeError("429：触发限流，需要退避后重试")
if resp.status >= 500:
raise RuntimeError(f"{resp.status}：服务端暂时故障，重试即可")
if resp.status != 200:
text = await resp.text()
raise RuntimeError(f"{resp.status}：未知错误 {text[:200]}")
if stream:
return await 解析流式响应(resp)
resp_json = await resp.json()
# 容错：服务端偶尔返回不完整 JSON 结构，逐层 .get 而不是裸下标
if "choices" not in resp_json:
raise RuntimeError(f"响应缺少 choices 字段：{str(resp_json)[:200]}")
return resp_json
except RuntimeError as e:
last_err = e
可重试 = "429" in str(e) or "500" in str(e) or "缺少 choices" in str(e)
print(f" ⚠️ {轮次说明}第 {attempt + 1} 次尝试失败：{e}")
if not 可重试 or attempt == 最大重试:
break
等待 = 2 ** attempt # 指数退避：1s → 2s → 4s
print(f" ⏳ {等待}s 后重试（指数退避）...")
await asyncio.sleep(等待)
except asyncio.TimeoutError:
last_err = RuntimeError("请求超时")
print(f" ⚠️ {轮次说明}第 {attempt + 1} 次尝试超时")
if attempt == 最大重试:
break
await asyncio.sleep(2 ** attempt)
except aiohttp.ClientError as e:
last_err = RuntimeError(f"网络错误：{type(e).__name__}")
print(f" ⚠️ {轮次说明}网络错误：{e}")
if attempt == 最大重试:
break
await asyncio.sleep(2 ** attempt)
raise last_err or RuntimeError("调用失败")
async def 解析流式响应(resp):
"""第 7 步：SSE 格式逐行解析——每个 chunk 是一行 data: {...}"""
full = ""
usage = {}
print(" 助手（流式）：", end="", flush=True)
async for raw in resp.content:
line = raw.decode("utf-8").strip()
if not line.startswith("data:"):
continue # SSE 还有 event:/retry: 等行，只关心 data:
data = line[len("data:"):].strip()
if data == "[DONE]":
break # 流结束的哨兵
try:
chunk = json.loads(data)
except json.JSONDecodeError:
continue # 半行/脏数据直接跳过，不让解析失败打断流
if chunk.get("usage"):
usage = chunk["usage"] # stream_options 包含 usage 时在最后的 chunk 里
delta = chunk.get("choices", [{}])[0].get("delta", {})
piece = delta.get("content") or ""
if piece:
full += piece
print(piece, end="", flush=True)
print()
return {"choices": [{"message": {"role": "assistant", "content": full}}], "usage": usage}
def 记账(usage):
usage_total["calls"] += 1
usage_total["prompt_tokens"] += usage.get("prompt_tokens", 0)
usage_total["completion_tokens"] += usage.get("completion_tokens", 0)
# ==================== 8 个步骤 ====================
def 第1步_key管理():
分隔线("第 1 步：API Key 的正确管理")
key = 找key()
if key:
print(f" ✅ 从环境变量读到 Key：{脱敏(key)}")
else:
print(" ⚠️ 未检测到 API Key（DEEPSEEK_API_KEY / OPENAI_API_KEY / LLM_API_KEY）")
print(" 💡 铁律：Key 只放环境变量；打印/报错一律脱敏；.env 必须进 .gitignore")
def 第2步_构造请求():
分隔线("第 2 步：构造第一个真实请求")
messages = [
{"role": "system", "content": "你是一个简洁的助手。"},
{"role": "user", "content": "用一句话介绍 Python 的 asyncio。"},
]
print(json.dumps(构造请求体(messages), ensure_ascii=False, indent=2))
print(" 💡 认识每个字段：model 选模型、messages 是全部上下文、")
print(" max_tokens 封顶输出、temperature 控制随机性")
def 第3步_离线解析演示():
print("\n【第 3 步（离线演示）：解析一个真实格式的响应】")
样例 = {
"choices": [{"message": {"role": "assistant", "content": "asyncio 是 Python 的异步 IO 框架。"}}],
"usage": {"prompt_tokens": 21, "completion_tokens": 12},
}
message, usage = 解析响应(样例)
print(f" 回答：{message['content']}")
print(f" 用量：prompt {usage['prompt_tokens']} + completion {usage['completion_tokens']} tokens")
记账(usage)
async def 第3步_发送请求(session):
分隔线("第 3 步：发送第一个真实请求")
messages = [
{"role": "system", "content": "你是一个简洁的助手。"},
{"role": "user", "content": "用一句话介绍 Python 的 asyncio。"},
]
resp_json = await call_llm(session, messages, 轮次说明="第3步 ")
message, usage = 解析响应(resp_json)
记账(usage)
print(f" ✅ 回答：{message['content']}")
print(f" 💰 用量：prompt {usage.get('prompt_tokens')} + completion {usage.get('completion_tokens')} tokens")
async def 第4步_错误处理(session):
分隔线("第 4 步：错误处理四件套（超时/状态码/重试/退避）")
print(" 上一条的调用已经带全了这四件套（见 call_llm 源码）：")
print(" ① asyncio.timeout(20) 每次请求都有超时")
print(" ② 状态码分诊：401 不重试（Key 错）、429/5xx 重试、其他报错")
print(" ③ 指数退避重试：1s → 2s，最多 2 次")
print(" ④ JSON/结构容错：响应缺字段时不裸下标崩溃")
print(" 现在故意用错误的 Key 调用，看 401 分诊：")
try:
async with asyncio.timeout(20):
async with session.post(
f"{BASE_URL}/chat/completions",
json=构造请求体([{"role": "user", "content": "hi"}]),
headers={"Authorization": "Bearer sk-wrong-key", "Content-Type": "application/json"},
) as resp:
if resp.status == 401:
print(" ✅ 收到 401——Key 无效时不重试，直接提示用户检查配置")
else:
print(f" ℹ️ 收到状态码 {resp.status}")
except Exception as e:
print(f" ℹ️ 网络不通（{type(e).__name__}）——线上环境这会被 call_llm 的重试兜住")
async def 第5步_多轮对话(session):
分隔线("第 5 步：多轮对话——上下文就是 messages 列表")
messages = [
{"role": "system", "content": "你是一个简洁的助手。"},
]
messages.append({"role": "user", "content": "我叫小张，我最喜欢的数字是 42。"})
resp = await call_llm(session, messages, max_tokens=256, 轮次说明="第5步-1 ")
记账(解析响应(resp)[1])
print(f" 助手：{解析响应(resp)[0]['content'][:60]}")
messages.append({"role": "assistant", "content": 解析响应(resp)[0]["content"]})
messages.append({"role": "user", "content": "我叫什么名字？最喜欢的数字是多少？"})
resp = await call_llm(session, messages, max_tokens=256, 轮次说明="第5步-2 ")
message, usage = 解析响应(resp)
记账(usage)
print(f" 助手：{message['content'][:80]}")
print(" 💡 答对了——因为历史里带着第一轮的 user 消息。上下文 = 你给它的列表")
TOOLS = [
{
"type": "function",
"function": {
"name": "calculator",
"description": "执行四则运算，输入四则运算表达式字符串",
"parameters": {
"type": "object",
"properties": {"expression": {"type": "string", "description": "如 '3 + 5 * 2'"}},
"required": ["expression"],
},
},
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的天气",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string", "description": "城市名"}},
"required": ["city"],
},
},
},
]
def 执行工具(name, arguments):
"""本地执行工具——和 Day 4/6 的注册表一个道理，这里直接 if/else"""
if name == "calculator":
try:
return str(eval(arguments["expression"], {"__builtins__": {}}, {}))
except Exception as e:
return f"错误: {e}"
if name == "get_weather":
return {"北京": "晴 25°C"}.get(arguments.get("city", ""), "天气未知")
return f"未知工具 {name}"
async def 第6步_FunctionCalling(session):
分隔线("第 6 步：真实 Function Calling（Day 4 的真实版）")
messages = [{"role": "user", "content": "帮我计算 3 + 5 * 2 等于多少？"}]
for 轮 in range(3): # 有上限，防费用失控
resp = await call_llm(session, messages, tools=TOOLS, max_tokens=512,
轮次说明=f"第6步-轮{轮 + 1} ")
message, usage = 解析响应(resp)
记账(usage)
if not message.get("tool_calls"):
print(f" ✅ 最终回答：{message['content']}")
return
messages.append(message) # assistant 的 tool_calls 也要回填历史
for tc in message["tool_calls"]:
name = tc["function"]["name"]
args = json.loads(tc["function"]["arguments"]) # arguments 是 JSON 字符串！
result = 执行工具(name, args)
print(f" 🔧 第 {轮 + 1} 轮 LLM 要求调用 {name}({args}) → {result}")
messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result})
print(" ⚠️ 达到最大轮数，停止（真实项目同样要设上限）")
async def 第7步_流式输出(session):
分隔线("第 7 步：流式输出——打字机效果的真实实现")
messages = [{"role": "user", "content": "用三句话介绍 AI Agent。"}]
body = 构造请求体(messages, stream=True, max_tokens=256)
body["stream_options"] = {"include_usage": True} # 让最后一块带 usage
headers = {"Authorization": f"Bearer {找key()}", "Content-Type": "application/json"}
async with asyncio.timeout(30):
async with session.post(f"{BASE_URL}/chat/completions", json=body,
headers=headers) as resp:
result = await 解析流式响应(resp)
记账(result["usage"])
print(" 💡 逐 chunk 到达就逐 chunk 打印——ChatGPT 的打字机效果就是这个原理")
def 第8步_成本汇总():
分隔线("第 8 步：成本统计——每次调用都要记账")
p, c = usage_total["prompt_tokens"], usage_total["completion_tokens"]
print(f" 本次教程共调用 {usage_total['calls']} 次 LLM")
print(f" Token：prompt {p} + completion {c} = {p + c}")
print(" 💡 生产做法：每次调用记录 usage 入库/日志，设日预算告警")
# ==================== 主流程 ====================
async def main():
print("=" * 60)
print("真实 LLM API 实战教程（tutorial_03）")
print("=" * 60)
第1步_key管理()
第2步_构造请求()
key = 找key()
if aiohttp is None:
print("\n❌ 缺少 aiohttp：pip install aiohttp （装好后重跑本教程）")
return
if not key:
第3步_离线解析演示()
第8步_成本汇总()
print("\n" + "=" * 60)
print("📌 以上是离线部分。配置环境变量 DEEPSEEK_API_KEY 后重新运行，")
print(" 即可体验第 3-7 步：真实调用 / 错误重试 / 多轮对话 /")
print(" Function Calling / 流式输出（有免费额度，本教程消耗很小）")
print("=" * 60)
return
timeout = aiohttp.ClientTimeout(total=None) # 每个请求内部已有 asyncio.timeout
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
await 第3步_发送请求(session)
await 第4步_错误处理(session)
await 第5步_多轮对话(session)
await 第6步_FunctionCalling(session)
await 第7步_流式输出(session)
except RuntimeError as e:
print(f"\n⚠️ API 调用中断：{e}")
print("💡 这正是第 4 步教的内容——检查 Key / 网络 / 额度后重跑即可")
except Exception as e:
print(f"\n⚠️ 网络异常：{type(e).__name__}: {e}")
print("💡 检查网络 / 代理后重跑；离线部分（第 1-3 步）不受影响")
第8步_成本汇总()
print("\n🎉 学完本教程，你已经具备把 Day 4-10 的 mock 全部换成真实 API 的能力")
print(" 下一步：Day 11-15 项目实战（见 index.html 路线图）")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n👋 已中断")