通过 Python 把 AI 大语言模型接入自己的项目
本文以开源项目 HuluAiChat 为例,说明如何用 Python 将任意「OpenAI 兼容」的 AI 聊天模型接入到自己的应用里。
Python 调用 AI 大语言模型主要基于 OpenAI 兼容 API 实现。通过安装 openai 库构造客户端,设置 base_url 和 api_key,即可对接各类模型服务。支持流式与非流式请求,利用迭代器处理 chunk 数据实现逐字输出。核心参数包括 model、messages、temperature 等,用于控制生成行为。消息格式遵循 system/user/assistant 角色定义,支持多轮对话。错误处理需捕获连接异常与状态码错误,生产环境建议增加重试机制。结合分层架构设计,可构建包含 UI、业务逻辑与基础设施的完整应用,实现本地历史存储与多模型切换。

本文以开源项目 HuluAiChat 为例,说明如何用 Python 将任意「OpenAI 兼容」的 AI 聊天模型接入到自己的应用里。
现成产品(如 ChatGPT 网页、各类 App)已经很好用,但在这些场景里你会希望把大模型能力嵌进自己的项目:
用 Python + OpenAI 兼容 API 可以很低成本地实现上述目标。下面以 「如何用 Python 调用 AI 聊天」 为核心,详解 openai 库的用法与参数,再简要结合 HuluChat 说明如何接到完整项目里。
只要你的 API 是 OpenAI 兼容 的(提供 POST /v1/chat/completions,请求/响应格式一致),用官方 openai Python 库即可,无需区分具体厂商。本节是全文重点:把客户端构造、create() 参数、消息格式、流式/非流式、错误处理等讲清楚,方便你直接用到自己的项目。
安装依赖:
pip install openai>=1.0.0
最小示例:构造客户端并发起一次流式请求。
from openai import OpenAI
client = OpenAI(
base_url="https://api.openai.com/v1", # 或你的 API 地址
api_key="your-api-key",
)
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "用一句话介绍 Python。"}],
stream=True,
)
for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
print()
把 base_url、api_key、model 换成你的环境,即可在任意 OpenAI 兼容服务上跑起来。
OpenAI() 参数说明OpenAI() 用于创建与 API 服务通信的客户端,常用参数如下。
| 参数 | 类型 | 说明 |
|---|---|---|
base_url | str | API 根地址,如 https://api.openai.com/v1。国内/自建可填 https://api.deepseek.com/v1、https://openai.xxx.com/v1 等。不填则默认官方 OpenAI。 |
api_key | str | 对应服务的密钥。部分兼容服务允许占位符(如 dummy),但生产环境必须填有效 key。 |
timeout | `float | None` |
max_retries | int | 失败时最大重试次数,默认 2。设为 0 可关闭自动重试。 |
http_client | 自定义 HTTP 客户端 | 可注入自己的 httpx.Client,用于代理、自定义 header 等。 |
示例:带超时与重试的客户端。
client = OpenAI(
base_url="https://api.openai.com/v1",
api_key="your-api-key",
timeout=120.0,
max_retries=2,
)
chat.completions.create(..., stream=True)流式调用时,接口返回的是一个迭代器,每次迭代得到一个「片段」对象,片段里当前新增的文本在 chunk.choices[0].delta.content。
要点:
stream=True 必须传入,否则返回的是整段完成后再返回的单个对象,而不是迭代器。chunk 可能有 chunk.choices[0].delta.content 为 None(例如只更新 role 或仅心跳),所以判断后再用。usage(部分服务在流结束前会发一个带 usage 的 chunk,需看具体实现)。示例:流式打印并拼接完整回复。
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "用三句话介绍 Python。"}],
stream=True,
)
full = []
for chunk in stream:
if chunk.choices and len(chunk.choices) > 0:
delta = chunk.choices[0].delta
if getattr(delta, "content", None):
full.append(delta.content)
print(delta.content, end="")
print()
reply = "".join(full)
create() 常用参数详解client.chat.completions.create() 的常用参数如下,兼容 OpenAI 的厂商大多支持其中大部分。
| 参数 | 类型 | 说明 |
|---|---|---|
model | str | 模型 ID,如 gpt-4o-mini、deepseek-chat、qwen-plus 等,由服务方规定。 |
messages | list[dict] | 对话历史,见 2.6 节。必填。 |
stream | bool | 是否流式返回。True 时返回迭代器;False 时返回一个完整 ChatCompletion 对象。 |
temperature | float | 采样随机度,范围一般 0~2。越高越随机,越低越确定。做代码/翻译时可设 0.2~0.3。 |
top_p | float | 核采样:只从概率质量前 top_p 的 token 中采样。与 temperature 二选一调即可。 |
max_tokens | int | 回复最大 token 数(部分服务用 max_completion_tokens)。不设则用模型默认上限。 |
stop | `str | list[str]` |
frequency_penalty | float | -2~2,正值减少重复,让模型少重复已出现过的词。 |
presence_penalty | float | -2~2,正值鼓励谈论新话题。 |
n | int | 同一条请求生成几条回复,默认 1。大于 1 时 choices 有多条,按需取用。 |
response_format | dict | 约束输出格式,如 {"type": "json_object"} 要求返回合法 JSON。 |
tools | list | 函数/工具列表,用于 Function Calling;与 tool_choice 配合。 |
seed | int | 固定随机种子,便于复现(部分模型支持)。 |
示例:带温度、最大 token 和 stop 的调用。
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "写一首四句诗,关于编程。"}],
stream=False,
temperature=0.8,
max_tokens=200,
stop=["。", "\n\n"],
)
text = resp.choices[0].message.content
当 **stream=False**(默认)时,create() 返回一个 ChatCompletion 对象,常用属性:
id:本次完成的 ID。**choices**:列表,通常只有一项。choices[0].message 是助手消息;choices[0].message.content 是文本内容。**choices[0].finish_reason**:结束原因,如 "stop"(正常结束)、"length"(达到 max_tokens)、"tool_calls"(调用了工具)等。**usage**:若服务返回则有 prompt_tokens、completion_tokens、total_tokens,用于计费或统计。示例:非流式并取完整回复与 token 数。
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "1+1=?"}],
stream=False,
)
msg = completion.choices[0].message
print(msg.content) # 助手回复文本
print(msg.role) # "assistant"
print(completion.choices[0].finish_reason) # "stop" 等
if completion.usage:
print(completion.usage.total_tokens) # 总 token 数
messages 与多轮对话messages 是按顺序的对话列表,每项为 {"role": "...", "content": "..."}。
| role | 说明 |
|---|---|
system | 系统提示,设定助手身份、风格、规则。可选,多数放在第一条。 |
user | 用户说的话。 |
assistant | 助手之前的回复,用于多轮对话时把历史带上。 |
多轮对话示例:带 system,并保留一轮 user/assistant 历史。
messages = [
{"role": "system", "content": "你是一个简洁的技术助手,只回答一句话。"},
{"role": "user", "content": "Python 的 GIL 是什么?"},
{"role": "assistant", "content": "GIL 是全局解释器锁,同一时刻只允许一个线程执行 Python 字节码。"},
{"role": "user", "content": "怎么规避?"},
]
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
stream=False,
)
print(completion.choices[0].message.content)
部分兼容 API 还支持 多模态:user 的 content 可以是数组,包含 {"type": "text", "text": "..."} 和 {"type": "image_url", "image_url": {"url": "..."}} 等,用于图文问答。
client.models.list() 可返回可用模型列表,便于在 UI 里展示或校验 model 是否有效。create() 里传入 tools 和 tool_choice,模型会返回 tool_calls,你本地执行对应函数后再把结果以 role="tool" 的消息 append 到 messages 并再次调用,实现「模型决定调哪个函数、你执行并回传」的流程。response_format={"type": "json_object"} 让模型尽量输出合法 JSON,便于解析。usage 返回 token 统计,需按各服务文档处理。这些在「最小接入」时可选;先把 create() 的 model / messages / stream 用熟,再按需加参数即可。
常见异常来自 openai 库:
openai.APIConnectionError:网络/连接问题,可视为临时错误,适合重试。**openai.APIStatusError**:HTTP 状态码非 2xx,如 401(key 错误)、429(限流)、5xx(服务端错误)。可通过 e.status_code、e.message 区分是否可重试(例如 5xx 可重试,401 则不必)。示例:捕获并区分错误。
from openai import OpenAI, APIConnectionError, APIStatusError
client = OpenAI(base_url="...", api_key="...")
try:
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Hi"}],
stream=True,
)
for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
except APIConnectionError as e:
print("连接失败,可重试:", e)
except APIStatusError as e:
print("API 错误:", e.status_code, e.message)
if e.status_code and 500 <= e.status_code < 600:
print("服务端错误,可稍后重试")
生产环境可在此基础上加指数退避重试、超时和日志(注意不要打印 api_key)。
小结第二节:用 Python 使用 AI 聊天 = 用 OpenAI(base_url=..., api_key=..., ...) 构造客户端,用 **client.chat.completions.create(model=..., messages=..., stream=True/False, ...)** 发请求;流式用迭代器取 chunk.choices[0].delta.content,非流式用 completion.choices[0].message.content。掌握 messages 格式和常用参数后,即可接入任意 OpenAI 兼容服务,并在此基础上做多模型、多轮对话和错误处理。
HuluAiChat 是一个轻量桌面 AI 聊天应用,用 Python + CustomTkinter 编写,底层只用 openai 库的兼容接口,会话与消息存 SQLite,配置在用户目录 config.json,支持多 Provider 切换、流式对话与亮/暗主题。下面用其架构说明如何把「最小示例」扩展成可维护的完整应用。
| 类别 | 技术 |
|---|---|
| 语言 | Python 3.10+ |
| 界面 | CustomTkinter |
| 聊天 API | openai 库(OpenAI 兼容) |
| 持久化 | SQLite |
| 配置 | 用户目录 config.json |
| 打包 | PyInstaller(Windows exe) |
HuluAiChat 采用清晰分层:UI 层(主窗口、设置)→ 应用层 AppService(会话、发消息、配置)→ 基础设施(ConfigStore、SessionRepo、MessageRepo、ChatClient)。「用 Python 接入 AI 聊天」落在 ChatClient:上层只依赖流式回调抽象,具体是否 OpenAI 兼容由 Chat 层封装。
基础设施:ChatClient(流式 API)、ConfigStore、SessionRepository、MessageRepository,均可替换。
从用户点击发送到界面逐字显示、再写入数据库:UI 将用户输入和 chunk_queue 交给 AppService;AppService 在后台线程调用 ChatClient.stream_chat(provider, messages, on_chunk),在 on_chunk 里把片段放入队列;主线程轮询队列更新 UI;收到 DoneChunk 后把完整助手回复写入 MessageRepository。你在自己项目里也可采用:队列 + 后台线程调流式 API + 主线程消费队列更新 UI。
HuluAiChat
src/chat/client.py 定义流式片段类型(TextChunk、DoneChunk、ChatError)和抽象 ChatClient.stream_chat(provider, messages, on_chunk);src/config/models.py 用 Provider(id、name、base_url、api_key、model_id)表示一个模型配置;OpenHuluChatClient 在 src/chat/openai_client.py 里用 OpenAI(base_url=..., api_key=...) 和 chat.completions.create(..., stream=True) 实现流式回调,错误统一转为 ChatError 经 on_chunk 回传。
这样任何 OpenAI 兼容或自研实现只要实现 ChatClient 即可接入。OpenAI(...) + chat.completions.create(..., stream=True);多模型可像 HuluChat 一样用 Provider 存多组 base_url / api_key / model_id。ChatClient 抽象和 stream_chat(provider, messages, on_chunk),当前用 OpenHuluChatClient,后续可换 Azure/自建实现并注入。queue.Queue() 给应用层并轮询;后台线程在 on_chunk 里 chunk_queue.put(chunk),参考 HuluChat 的 AppService.send_message。pip install openai>=1.0.0;配置用环境变量、JSON 或数据库均可。ChatClient 并注入。openai,用 OpenAI(base_url=..., api_key=...) + chat.completions.create(..., stream=True/False, ...) 即可对接任意 OpenAI 兼容 服务;第二节已展开各参数、消息格式、流式/非流式与错误处理。ChatClient 抽象,便于复用或替换实现。
微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML转Markdown 互为补充。 在线工具,Markdown转HTML在线工具,online