Python 调用 AI 接口

用 Python requests 调用 OpenRouter API 实现流式对话。

#type / howto #status / growing #tech / lang / python #resource / python #tech / ai

[!info] related notes

前置笔记: 免费ai服务, Python环境配置

所属 MOC: 学习 AI Agent MOC

相关 MOC: Python 大模型应用开发 MOC

相关概念: 大语言模型, Agent, 服务端推送 (SSE)

相关资源: OpenRouter

Python 调用 AI 接口

前置条件

了解基础概念

配置基础环境

通过免费ai服务找到一个合适的 AI 服务提供商，下面以 OpenRouter 为例
配置好 Python 环境：Python环境配置

核心理解

Agent 不是魔法，本质是：

LLM + instructions + tools + loop

先通过 AI 提供商的接口来制作一个最小 Agent。

实战：流式调用 OpenRouter API

紧凑版

import requests
import json

question = "How would you build the tallest building ever?"

url = "https://openrouter.ai/api/v1/chat/completions"
headers = {
  "Authorization": f"Bearer <OPENROUTER_API_KEY>",
  "Content-Type": "application/json"
}

payload = {
  "model": "openai/gpt-4o",
  "messages": [{"role": "user", "content": question}],
  "stream": True
}

buffer = ""
with requests.post(url, headers=headers, json=payload, stream=True) as r:
  for chunk in r.iter_content(chunk_size=1024, decode_unicode=True):
    buffer += chunk
    while True:
      try:
        line_end = buffer.find('\n')
        if line_end == -1:
          break

        line = buffer[:line_end].strip()
        buffer = buffer[line_end + 1:]

        if line.startswith('data: '):
          data = line[6:]
          if data == '[DONE]':
            break

          try:
            data_obj = json.loads(data)
            content = data_obj["choices"][0]["delta"].get("content")
            if content:
              print(content, end="", flush=True)
          except json.JSONDecodeError:
            pass
      except Exception:
        break

逐行注释版

[!note]- 展开查看完整注释版本

# 导入 requests 库
# requests 是 Python 里常用的 HTTP 请求库
# 我们可以用它向 OpenRouter 的 API 发送请求
import requests

# 导入 json 库
# json 库可以把 JSON 字符串解析成 Python 字典
# OpenRouter 返回的数据就是 JSON 格式，所以后面需要用到它
import json


# 定义一个问题字符串，并把它赋值给 question 变量
# 这里的问题是：How would you build the tallest building ever?
# 也就是"你会如何建造有史以来最高的建筑？"
# 后面我们会把这个问题发送给 AI 模型
question = "How would you build the tallest building ever?"


# 定义 OpenRouter 的聊天补全 API 地址
# 这个 URL 是 OpenRouter 官方提供的接口地址
# 我们会把请求发送到这个地址
url = "https://openrouter.ai/api/v1/chat/completions"


# 定义请求头 headers
# headers 用来告诉服务器：
# 1. 我是谁，也就是 Authorization 认证信息
# 2. 我发送的数据格式是什么，也就是 Content-Type
headers = {
  # Authorization 是身份认证字段
  # Bearer 后面要放你的 OpenRouter API Key
  # <OPENROUTER_API_KEY> 只是占位符
  # 实际使用时，需要替换成你自己的 API Key
  #
  # f"Bearer <OPENROUTER_API_KEY>" 是 Python 的 f-string 写法
  # f-string 可以在字符串中插入变量
  # 不过这里暂时没有插入变量，只是写了一个固定字符串
  "Authorization": f"Bearer <OPENROUTER_API_KEY>",

  # Content-Type 表示请求体的数据格式
  # application/json 的意思是：
  # 我们发送给服务器的数据是 JSON 格式
  "Content-Type": "application/json"
}


# 定义请求体 payload
# payload 是我们要发送给 OpenRouter API 的核心数据
# 它会告诉 OpenRouter：
# 1. 使用哪个模型
# 2. 给模型发送什么消息
# 3. 是否使用流式输出
payload = {
  # model 指定要调用的模型
  # 这里使用的是 openai/gpt-4o
  # 在 OpenRouter 里，模型名称通常是 "提供商/模型名" 的格式
  "model": "openai/gpt-4o",

  # messages 是对话消息列表
  # Chat Completions API 通常使用 messages 来表示上下文
  # 它是一个列表，列表里每一项都是一条消息
  "messages": [
    {
      # role 表示消息角色
      # user 表示这条消息是用户发给 AI 的
      "role": "user",

      # content 表示消息内容
      # 这里把前面定义的 question 变量作为用户问题发送给模型
      "content": question
    }
  ],

  # stream 设置为 True，表示使用流式输出
  # 流式输出的意思是：
  # AI 不是等完整回答生成完再一次性返回
  # 而是一边生成，一边把内容分片返回
  # 这样用户可以更快看到模型正在输出的内容
  "stream": True
}


# 定义一个空字符串 buffer
# buffer 的作用是临时存放服务器返回的数据片段
# 因为流式返回的数据可能不是一次返回一整行
# 有可能一条完整数据被拆成多个 chunk
# 所以我们需要用 buffer 把碎片先拼起来
buffer = ""


# 使用 requests.post 发送 POST 请求
# POST 请求一般用于向服务器提交数据
# 这里我们向 OpenRouter 的 chat completions 接口发送请求
#
# 参数说明：
# - url：请求的 API 地址
# - headers=headers：把前面定义的请求头传进去
# - json=payload：把 payload 作为 JSON 请求体发送
# - stream=True：告诉 requests 不要一次性下载完整响应，而是按流读取
#
# with ... as r:
# 这是 Python 的上下文管理器写法
# 它可以保证请求结束后自动关闭连接
with requests.post(url, headers=headers, json=payload, stream=True) as r:

  # 遍历服务器返回的流式数据
  # iter_content 会不断读取响应内容
  #
  # chunk_size=1024 表示每次最多读取 1024 字节
  # decode_unicode=True 表示尽量把字节数据解码成字符串
  for chunk in r.iter_content(chunk_size=1024, decode_unicode=True):

    # 把当前收到的数据片段追加到 buffer 后面
    # 因为一个完整的 SSE 数据行可能会被拆成多个 chunk
    # 所以不能直接处理 chunk，而是先放进 buffer 里
    buffer += chunk

    # 使用 while True 不断尝试从 buffer 里取出完整的一行
    # 如果 buffer 里有多行数据，就会在这个循环里一行一行处理
    while True:

      try:
        # 在 buffer 中查找第一个换行符 \n 的位置
        # SSE 协议里的数据通常是一行一行传输的
        # 所以我们要根据换行符判断一行是否完整
        line_end = buffer.find('\n')

        # 如果没有找到换行符，说明当前 buffer 里还没有完整的一行
        # 这时先跳出 while 循环，继续等待下一个 chunk
        if line_end == -1:
          break

        # 取出 buffer 中第一行的内容
        # buffer[:line_end] 表示从开头截取到换行符之前
        # strip() 用来去掉前后的空格、换行符等
        line = buffer[:line_end].strip()

        # 把已经处理过的这一行从 buffer 中删除
        # buffer[line_end + 1:] 表示保留换行符后面的内容
        # 因为后面可能还有其他未处理的数据
        buffer = buffer[line_end + 1:]

        # 判断这一行是不是以 data: 开头
        # OpenRouter 的流式响应使用 SSE 格式
        # SSE 中真正的数据行通常长这样：
        # data: {"choices":[...]}
        if line.startswith('data: '):

          # 去掉前面的 "data: "，只保留真正的 JSON 字符串
          # line[6:] 的意思是从第 6 个字符之后开始截取
          # 因为 "data: " 刚好长度是 6
          data = line[6:]

          # 如果 data 是 [DONE]，说明流式输出结束了
          # OpenAI / OpenRouter 风格的流式 API 通常用 [DONE] 表示结束
          if data == '[DONE]':
            break

          # 尝试解析 JSON 数据
          try:
            # 把 JSON 字符串转换成 Python 字典
            # 例如把 '{"a": 1}' 转成 {"a": 1}
            data_obj = json.loads(data)

            # 从返回数据中取出模型本次增量输出的文本
            #
            # data_obj["choices"] 表示取 choices 字段
            # choices 通常是一个列表
            #
            # data_obj["choices"][0] 表示取第一个候选回答
            #
            # ["delta"] 表示取这次流式返回的增量内容
            #
            # .get("content") 表示尝试获取 content 字段
            # 用 get 的好处是：
            # 如果 content 不存在，不会直接报错，而是返回 None
            content = data_obj["choices"][0]["delta"].get("content")

            # 如果 content 有内容，就打印出来
            # 有些流式片段可能不包含 content，比如只包含角色信息或结束信息
            if content:
              # print 默认会在末尾加换行
              # end="" 表示不要自动换行
              #
              # flush=True 表示立刻把内容输出到终端
              # 这样就能看到 AI 一边生成一边显示
              print(content, end="", flush=True)

          # 如果 JSON 解析失败，就忽略这条数据
          # 这通常是因为数据不完整或者不是合法 JSON
          # 这里用 pass 表示什么都不做
          except json.JSONDecodeError:
            pass

      # 捕获其他异常
      # 如果处理过程中出现任何错误，就跳出 while 循环
      # 注意：真实项目中不建议简单地 except Exception 后直接 break
      # 更好的做法是打印错误信息，方便调试
      except Exception:
        break

关键概念速查

概念	说明
`requests.post()`	发送 HTTP POST 请求
`stream=True`	流式读取响应，不等全部下载完
`iter_content()`	逐块读取响应内容
`json.loads()`	把 JSON 字符串解析成 Python 字典
`data: [DONE]`	SSE 流式输出结束标记
`flush=True`	立即输出到终端，不等缓冲区满

常见问题

ModuleNotFoundError: No module named 'requests'：需要先 pip install requests
401 Unauthorized：API Key 未设置或已过期
KeyError: 'choices'：返回数据格式不符合预期，检查模型名称是否正确
中文乱码：确保终端编码为 UTF-8