通过 Python 把 AI 大语言模型接入自己的项目
现成产品（如 ChatGPT 网页、各类 App）已经很好用，但在这些场景里你会希望把大模型能力嵌进自己的项目：
PythonAI算法
使用 Python 接入 AI 大语言模型至个人项目
基于 Python 的 openai 库实现 AI 大语言模型接入个人项目的技术方案。详解客户端构造、流式与非流式调用参数、消息格式及错误处理。结合分层架构示例，演示如何将最小代码扩展为支持多模型、本地存储及后台线程调用的完整应用。
基于 Python 的 openai 库实现 AI 大语言模型接入个人项目的技术方案。详解客户端构造、流式与非流式调用参数、消息格式及错误处理。结合分层架构示例，演示如何将最小代码扩展为支持多模型、本地存储及后台线程调用的完整应用。
现成产品（如 ChatGPT 网页、各类 App）已经很好用，但在这些场景里你会希望把大模型能力嵌进自己的项目：
用 Python + OpenAI 兼容 API 可以很低成本地实现上述目标。下面以 「如何用 Python 调用 AI 聊天」 为核心，详解 openai 库的用法与参数，再简要结合示例说明如何接到完整项目里。
现成产品（如 ChatGPT 网页、各类 App）已经很好用，但在这些场景里你会希望把大模型能力嵌进自己的项目：
用 Python + OpenAI 兼容 API 可以很低成本地实现上述目标。下面以 「如何用 Python 调用 AI 聊天」 为核心，详解 openai 库的用法与参数，再简要结合示例说明如何接到完整项目里。
只要你的 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 兼容服务，并在此基础上做多模型、多轮对话和错误处理。
这是一个轻量桌面 AI 聊天应用，用 Python + CustomTkinter 编写，底层只用 openai 库的兼容接口，会话与消息存 SQLite，配置在用户目录 config.json，支持多 Provider 切换、流式对话与亮/暗主题。下面用其架构说明如何把「最小示例」扩展成可维护的完整应用。
| 类别 | 技术 |
|---|---|
| 语言 | Python 3.10+ |
| 界面 | CustomTkinter |
| 聊天 API | openai 库（OpenAI 兼容） |
| 持久化 | SQLite |
| 配置 | 用户目录 config.json |
| 打包 | PyInstaller（Windows exe） |
示例项目采用清晰分层：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。
示例项目
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)；多模型可像示例一样用 Provider 存多组 base_url / api_key / model_id。ChatClient 抽象和 stream_chat(provider, messages, on_chunk)，当前用 OpenHuluChatClient，后续可换 Azure/自建实现并注入。queue.Queue() 给应用层并轮询；后台线程在 on_chunk 里 chunk_queue.put(chunk)，参考示例的 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
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online