# LLM API 实战:用 Python 构建智能问答系统
大语言模型火遍全球已经有两年了。GPT、Claude、通义千问这些名字你应该都听过。这些模型能理解自然语言,能生成内容,用来作客服、知识库问答、内容创作都很合适。
直接调 API 是最省事的办法。发个 HTTP 请求就能拿到回复,本文以 OpenAI 兼容 API 为例,手把手教你从零搭一个问答系统。
## 问题描述
很多开发者想用 LLM API,但不知道该怎么上手:
– HTTP 请求该怎么发?
– 多轮对话怎么做?
– 流式输出怎么实现?
– 想让模型按特定方式回答怎么控制?
下面一个个来解决。
## 详细步骤
—
安装依赖:
“`bash
pip install requests sseclient-py
“`
### 基础调用
先写一个最基础的例子,不过脑子就能跑:
“`python
import requests
import json
def chat_once(prompt, user_message):
“””单轮对话示例”””
url = “https://api.openai.com/v1/chat/completions”
headers = {
“Authorization”: f”Bearer {your_api_key}”,
“Content-Type”: “application/json”
}
data = {
“model”: “gpt-3.5-turbo”,
“messages”: [
{“role”: “system”, “content”: prompt},
{“role”: “user”, “content”: user_message}
],
“temperature”: 0.7,
“max_tokens”: 1000
}
response = requests.post(url, headers=headers, json=data)
return response.json()[“choices”][0][“message”][“content”]
“`
### 流式输出
想看到模型边想边回复?流式输出就是这个效果,体验很像跟真人聊天:
“`python
import requests
def chat_stream(prompt, user_message):
“””流式对话示例”””
url = “https://api.openai.com/v1/chat/completions”
headers = {
“Authorization”: f”Bearer {your_api_key}”,
“Content-Type”: “application/json”
}
data = {
“model”: “gpt-3.5-turbo”,
“messages”: [
{“role”: “system”, “content”: prompt},
{“role”: “user”, “content”: user_message}
],
“temperature”: 0.7,
“stream”: True
}
response = requests.post(url, headers=headers, json=data, stream=True)
for line in response.iter_lines():
if line:
line = line.decode(“utf-8”)
if line.startswith(“data: “):
data_str = line[6:]
if data_str == “[DONE]”:
break
chunk = json.loads(data_str)
if “choices” in chunk and chunk[“choices”]:
delta = chunk[“choices”][0].get(“delta”, {})
content = delta.get(“content”, “”)
if content:
print(content, end=””, flush=True)
“`
### 多轮对话
连续对话需要自己存历史消息:
“`python
class Chatbot:
def __init__(self, system_prompt, api_key, model=”gpt-3.5-turbo”):
self.api_key = api_key
self.model = model
self.messages = [{“role”: “system”, “content”: system_prompt}]
def ask(self, user_message):
“””发送消息并获取回复”””
self.messages.append({“role”: “user”, “content”: user_message})
url = “https://api.openai.com/v1/chat/completions”
headers = {
“Authorization”: f”Bearer {self.api_key}”,
“Content-Type”: “application/json”
}
data = {
“model”: self.model,
“messages”: self.messages,
“temperature”: 0.7
}
response = requests.post(url, headers=headers, json=data)
assistant_reply = response.json()[“choices”][0][“message”][“content”]
self.messages.append({“role”: “assistant”, “content”: assistant_reply})
return assistant_reply
def reset(self):
“””清空对话历史”””
self.messages = self.messages[:1]
“`
### 自定义控制
改 system prompt 就能让模型按你的方式回答:
“`python
# 法律助手模式
legal_prompt = “””你是一个专业的法律助手。
用通俗易懂的语言解释法律术语。
在回答结束时,添加一句风险提示:
“以上仅供参考,不构成法律意见,建议咨询专业律师。” “””
# 技术support模式
support_prompt = “””你是一个耐心的技术支持工程师。
始终保持友好语气。
如果用户的问题无法确定,引导用户提供更多细节。”””
# 创建不同的机器人实例
legal_bot = Chatbot(legal_prompt, api_key)
support_bot = Chatbot(support_prompt, api_key)
“`
## 完整代码
整合一下,就是一个能用的问答系统:
“`python
import requests
import json
import os
class SmartQA:
def __init__(self, api_key, system_prompt=”你是一个有用的助手”):
self.api_key = api_key
self.system_prompt = system_prompt
self.messages = []
self.history_limit = 10
def _build_messages(self, user_input):
messages = [{“role”: “system”, “content”: self.system_prompt}]
for role, content in self.messages[-self.history_limit:]:
messages.append({“role”: role, “content”: content})
messages.append({“role”: “user”, “content”: user_input})
return messages
def ask(self, user_input, stream=False):
messages = self._build_messages(user_input)
url = “https://api.openai.com/v1/chat/completions”
headers = {
“Authorization”: f”Bearer {self.api_key}”,
“Content-Type”: “application/json”
}
data = {
“model”: “gpt-3.5-turbo”,
“messages”: messages,
“temperature”: 0.7,
“max_tokens”: 1000,
“stream”: stream
}
if stream:
return self._stream_request(url, headers, data)
else:
return self._normal_request(url, headers, data)
def _normal_request(self, url, headers, data):
response = requests.post(url, headers=headers, json=data)
result = response.json()[“choices”][0][“message”][“content”]
self.messages.append((“user”, data[“messages”][-1][“content”]))
self.messages.append((“assistant”, result))
return result
def _stream_request(self, url, headers, data):
response = requests.post(url, headers=headers, json=data, stream=True)
full_content = “”
for line in response.iter_lines():
if line:
line = line.decode(“utf-8”)
if line.startswith(“data: “):
data_str = line[6:]
if data_str == “[DONE]”:
break
chunk = json.loads(data_str)
if “choices” in chunk and chunk[“choices”]:
delta = chunk[“choices”][0].get(“delta”, {})
content = delta.get(“content”, “”)
if content:
full_content += content
print(content, end=””, flush=True)
self.messages.append((“user”, data[“messages”][-1][“content”]))
self.messages.append((“assistant”, full_content))
return full_content
def clear_history(self):
self.messages = []
# 使用示例
if __name__ == “__main__”:
api_key = os.environ.get(“OPENAI_API_KEY”, “your-api-key-here”)
qa = SmartQA(
api_key=api_key,
system_prompt=”你是一个编程助手,用简洁清晰的代码示例帮助用户解决问题。”
)
# 测试对话
print(“用户:如何用 Python 读取文件?”)
print(“助手:”, end=””)
qa.ask(“如何用 Python 读取文件?”)
print(“\n”)
print(“用户:如何处理异常?”)
print(“助手:”, end=””)
qa.ask(“如何处理异常?”)
print(“\n”)
“`
## 运行结果
跑一下试试:
“`
用户:如何用 Python 读取文件?
助手:可以用以下几种方式:
1. 一次性读取整个文件:
with open(“file.txt”, “r”) as f:
content = f.read()
print(content)
2. 逐行读取:
with open(“file.txt”, “r”) as f:
for line in f:
print(line.strip())
3. 使用 pathlib(Python 3.4+):
from pathlib import Path
content = Path(“file.txt”).read_text()
推荐用 with 语句,自动关文件。
用户:如何处理异常?
助手:用 try-except 语句:
try:
result = 10 / 0
except ZeroDivisionError:
print(“不能除以零”)
except Exception as e:
print(f”发生错误: {e}”)
else:
print(“执行成功”)
finally:
print(“总是执行”)
可以捕获多个异常,每个 except 处理一种类型。
“`
## 总结
核心知识点就这几个:
– **基础 API 调用**:用 requests 库发 HTTP 请求
– **流式输出**:SSE 逐字显示,体验好
– **上下文管理**:自己存对话历史
– **自定义控制**:改 system prompt
这几个都搞清楚,就能搭自己的问答系统了。再往上学,可以加向量数据库做知识库,或者集成多个 LLM 做路由。
个人开发者建议先跑通最小版本,验证需求后再加功能。不是什么功能都要往里塞,够用就行。