关于GPT应用的几点记录
架构图画得再漂亮，一进项目还是会被 Token 账单和上下文窗口教做人。
第一次碰 OpenAI API 是 2023 年初。当时想法很天真：前端加个输入框，后端调 API，把答案丢回去，就算"完成 AI 改造"。
最早版本：一个对话脚本
那时候的代码大概长这样：
from openai import OpenAI
client = OpenAI(api_key="sk-xxx")
def chat(message):
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "你是一个有用的助手。"},
{"role": "user", "content": message}
]
)
return response.choices[0].message.content
print(chat("帮我写一个 Python 函数反转列表"))
这个版本能对话，但问题很多：
- 没记忆，每次对话都是独立的
- 没结构化输出，想提取内容还得自己解析
- 没错误处理，网络异常、超时、限流都是直接挂
- 没成本控制，用户聊多了 Token 费用不可控
第一个踩的坑是 Token 计费。有个测试用户连续问了十几个复杂问题，测试账号几分钟就烧了 2 美元。后来加上 Token 计数和限制：
import tiktoken
def count_tokens(text, model="gpt-3.5-turbo"):
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))
def chat_with_limit(message, max_tokens=1000):
input_tokens = count_tokens(message)
if input_tokens > max_tokens:
return "问题太长，请简化后重试。"
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "system", "content": "你是一个有用的助手。"},
{"role": "user", "content": message}],
max_tokens=max_tokens - input_tokens
)
return response.choices[0].message.content
加上记忆：从对话到上下文
要让它"记住"之前的对话，最简单的做法是把历史消息都塞进去：
class ChatSession:
def __init__(self):
self.messages = [
{"role": "system", "content": "你是一个有用的助手。"}
]
def chat(self, message):
self.messages.append({"role": "user", "content": message})
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=self.messages
)
answer = response.choices[0].message.content
self.messages.append({"role": "assistant", "content": answer})
return answer
这个版本跑了一段时间，问题又来了：对话一长，Token 就爆了，而且无关的历史对话会干扰当前问题。
后来改用滑动窗口，只保留最近 N 条：
from collections import deque
class ChatSessionWithWindow:
def __init__(self, window_size=6):
self.messages = deque(maxlen=window_size)
self.messages.append(
{"role": "system", "content": "你是一个有用的助手。"}
)
def chat(self, message):
self.messages.append({"role": "user", "content": message})
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=list(self.messages)
)
answer = response.choices[0].message.content
self.messages.append({"role": "assistant", "content": answer})
return answer
但这样还是不够。真正的问题是：什么内容需要记住？什么可以丢掉？如果用户在第三轮说了"我需要 Python 相关的回答"，第十轮又问"怎么反转数组"，这时候系统提示早就被挤出窗口了。
后来试了几种方案：
- 分层记忆：系统提示+长期记忆+短期对话
- 关键信息提取：每轮对话后自动提取关键信息
- 用户显式控制：允许用户标记"记住这个"或"忘掉刚才的"
最后在项目里实际用的是简化版分层：
class MemoryAssistant:
def __init__(self):
self.system_prompt = "你是一个专业的编程助手。"
self.long_term_memory = {} # key: user_id, value: summary
self.conversation_history = {} # key: session_id, value: deque
def get_system_messages(self, user_id):
messages = [{"role": "system", "content": self.system_prompt}]
if user_id in self.long_term_memory:
messages.append({
"role": "system",
"content": f"用户背景：{self.long_term_memory[user_id]}"
})
return messages
def chat(self, user_id, session_id, message):
# 获取或初始化会话历史
if session_id not in self.conversation_history:
self.conversation_history[session_id] = deque(maxlen=6)
# 构建完整消息
messages = self.get_system_messages(user_id)
messages.extend(list(self.conversation_history[session_id]))
messages.append({"role": "user", "content": message})
response = client.chat.completions.create(
model="gpt-4",
messages=messages
)
answer = response.choices[0].message.content
# 更新对话历史
self.conversation_history[session_id].append(
{"role": "user", "content": message}
)
self.conversation_history[session_id].append(
{"role": "assistant", "content": answer}
)
return answer
结构化输出：从文本到 JSON
初期为了让 AI 返回结构化数据，我在 Prompt 里写"请返回 JSON 格式"，结果它经常返回"这是 JSON 格式的回答：{…}"，或者字段名不对、类型不匹配。
后来发现 OpenAI 有专门的 JSON 模式：
def get_weather_info(location):
response = client.chat.completions.create(
model="gpt-4",
messages=[
{
"role": "system",
"content": "你是一个天气信息提取助手。从用户输入中提取地点信息。"
},
{"role": "user", "content": location}
],
response_format={"type": "json_object"}
)
return response.choices[0].message.content
但 JSON 模式只是保证格式，不保证内容符合预期。更可靠的方案是加上 Pydantic 模型：
from pydantic import BaseModel
from typing import Optional
class WeatherQuery(BaseModel):
location: str
date: Optional[str] = None
temperature_unit: str = "celsius"
def extract_weather_query(text: str) -> WeatherQuery:
schema = WeatherQuery.model_json_schema()
prompt = f"""
从以下文本中提取天气查询信息，返回符合以下 JSON Schema 的结果：
{schema}
文本：{text}
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一个精确的信息提取助手。"},
{"role": "user", "content": prompt}
],
response_format={"type": "json_object"}
)
return WeatherQuery.model_validate_json(
response.choices[0].message.content
)
用了一段时间后发现，简单场景下用 JSON 模式就够了，复杂场景还是得在 Prompt 里加示例。有个项目需要从技术文档里提取 API 定义，JSON 模式经常返回字段缺失，后来在 Prompt 里加了 2-3 个例子，准确率明显提升。
从对话到工具调用
2023 年底 OpenAI 推出 Function Calling（后来改名成 Tools），这个问题才算有了一个标准方案。之前我试过在 Prompt 里教 AI"如果用户问天气，就返回 WEATHER:地点"，但稳定性很差，经常该调用时不调用，不该调用时又瞎调用。
用 Function Calling 的版本：
import json
def get_current_weather(location, unit="celsius"):
# 这里应该是实际的天气 API 调用
return f"{location} 当前温度 25{unit[0]}"
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "获取指定地点的当前天气",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称，例如：北京、上海"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["location"]
}
}
}
]
def run_conversation():
messages = [
{"role": "user", "content": "北京今天天气怎么样？"}
]
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools
)
response_message = response.choices[0].message
tool_calls = response_message.tool_calls
if tool_calls:
# 执行工具调用
for tool_call in tool_calls:
if tool_call.function.name == "get_current_weather":
function_args = json.loads(tool_call.function.arguments)
weather_info = get_current_weather(
function_args.get("location"),
function_args.get("unit", "celsius")
)
messages.append(response_message)
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": tool_call.function.name,
"content": weather_info
})
# 获取最终回答
second_response = client.chat.completions.create(
model="gpt-4",
messages=messages
)
return second_response.choices[0].message.content
print(run_conversation())
但实际项目中，工具调用不是这么简单。几个常见问题：
工具定义不规范：description 写得不清楚，AI 就不知道什么时候该调用；参数定义太松，传进来的值就各种离谱。
错误处理：工具调用可能失败（网络问题、API 限流、参数错误），这时候要告诉 AI 出了什么错，让它决定是重试还是改方案。
并发调用：如果用户的问题需要查多个独立信息（比如同时查多个城市的天气），串行调用就慢了。
实际项目里的版本会复杂一些：
from concurrent.futures import ThreadPoolExecutor
from typing import Dict, Any
class ToolExecutor:
def __init__(self):
self.tools = {} # name -> function
def register(self, name: str, description: str, parameters: Dict[str, Any]):
def decorator(func):
self.tools[name] = {
"function": func,
"description": description,
"parameters": parameters
}
return func
return decorator
def to_openai_format(self):
return [
{
"type": "function",
"function": {
"name": name,
"description": tool["description"],
"parameters": tool["parameters"]
}
}
for name, tool in self.tools.items()
]
def execute(self, tool_call) -> str:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
try:
result = self.tools[name]["function"](**args)
return json.dumps({"success": True, "result": result})
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
executor = ToolExecutor()
@executor.register(
name="get_current_weather",
description="获取指定地点的当前天气",
parameters={
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称，例如：北京、上海"
}
},
"required": ["location"]
}
)
def get_current_weather(location: str) -> Dict[str, Any]:
# 实际项目里这里会是真实的 API 调用
return {"location": location, "temperature": 25, "condition": "晴"}
def run_agent_conversation(user_message: str) -> str:
messages = [{"role": "user", "content": user_message}]
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=executor.to_openai_format()
)
response_message = response.choices[0].message
while response_message.tool_calls:
messages.append(response_message)
# 并发执行所有工具调用
with ThreadPoolExecutor() as executor_pool:
for tool_call in response_message.tool_calls:
executor_pool.submit(
lambda tc: messages.append({
"tool_call_id": tc.id,
"role": "tool",
"name": tc.function.name,
"content": executor.execute(tc)
}),
tool_call
)
# 再次请求，获取下一步
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=executor.to_openai_format()
)
response_message = response.choices[0].message
return response_message.content
Prompt 优化：从模糊到精确
刚开始写 Prompt 时很随意，“你是一个有用的助手”、“帮我写一段代码"这种写法很常见。后来发现 Prompt 的质量直接影响结果，尤其是工具调用场景。
几个关键经验：
身份定义要具体：不要说"你是一个编程助手”，要说"你是一个专注于 Python 后端开发的工程师，熟悉 FastAPI、SQLAlchemy 和 PostgreSQL"。
约束条件要具体：除了"别生成错误代码"，还要写清楚"不确定 API 用法就说不知道，别编"。
输出格式要示例：尤其是结构化输出，给 1-2 个完整例子比写一堆说明管用。
实际项目里常用的 Prompt 模板：
SYSTEM_PROMPT_TEMPLATE = """
你是一个专业的{role}，专注于{expertise_area}。
## 你的职责
{responsibilities}
## 回答要求
1. 如果信息不足，明确提问而不是猜测
2. 如果不确定某个细节，标注出来
3. 代码示例要完整可运行，包含必要的 import 和错误处理
4. 解释技术选择时说明理由
## 不可做
- 不要编造不存在的库或 API
- 不要给出看起来正确但实际无法运行的代码
- 不要忽略用户明确提到的限制条件
{additional_constraints}
"""
def get_system_prompt(role="后端工程师", expertise_area="Python Web 开发"):
return SYSTEM_PROMPT_TEMPLATE.format(
role=role,
expertise_area=expertise_area,
responsibilities="\n".join([
"提供清晰、准确的解决方案",
"考虑性能、安全性和可维护性",
"给出完整可用的代码示例"
]),
additional_constraints="所有代码示例使用 Python 3.11+ 语法。"
)
但 Prompt 不是越长越好。后来发现，核心还是"描述清楚你想要什么"，而不是堆一堆规则。有时候一段简洁的例子比十条规则管用。
错误处理：从裸调到容错
最早的代码基本没错误处理，API 一报错前端就白屏。后来慢慢加上重试、降级、兜底。
几个关键点：
- 网络问题：超时、连接失败要重试，但要有限次数和退避策略。
- API 限流：OpenAI 有速率限制，达到限制要等一段时间。
- 内容安全：可能触发内容审核，这时候要优雅降级。
- 模型故障：偶尔会遇到模型返回异常格式或空内容。
实际项目里的版本：
import time
from tenacity import retry, stop_after_attempt, wait_exponential
from openai import RateLimitError, APITimeoutError
class RobustGPTClient:
def __init__(self, api_key: str, max_retries=3):
self.client = OpenAI(api_key=api_key)
self.max_retries = max_retries
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def chat_completion(self, messages, model="gpt-4"):
try:
response = self.client.chat.completions.create(
model=model,
messages=messages,
timeout=30 # 设置超时
)
return response.choices[0].message.content
except RateLimitError as e:
# 速率限制，等待后重试
wait_time = int(e.response.headers.get('retry-after', 5))
time.sleep(wait_time)
raise # 触发重试
except APITimeoutError:
# 超时，直接重试
raise
except Exception as e:
# 其他错误，记录并返回兜底响应
print(f"API 调用失败: {e}")
return "抱歉，我现在无法回答，请稍后再试。"
def safe_chat(self, messages, fallback_response="我遇到一些问题，无法继续。"):
try:
return self.chat_completion(messages)
except Exception as e:
print(f"所有重试均失败: {e}")
return fallback_response
还有一类错误是"模型胡说八道"，这个没法靠代码完全避免，但可以在设计上降低影响：比如让它标明不确定的内容，或者对于关键结果加人工审核。
从聊天机器人到智能助手
做到这一步，能对话、有记忆、能调用工具、能处理错误，基本上算一个"能干活"的智能助手了。但实际项目里，还有几个关键区别：
- 任务分解：复杂任务需要分解成多个步骤，每个步骤可能需要不同的工具。
- 状态管理：长时间运行的任务需要保存中间状态，支持恢复和回滚。
- 用户反馈：允许用户对结果进行修正，形成闭环。
这些能力就超出单纯的"对话"范畴了，接近于"Agent"的领域。我摸索了一个简化版的任务框架：
from enum import Enum
from dataclasses import dataclass
from typing import List, Optional
class TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class TaskStep:
id: str
description: str
status: TaskStatus = TaskStatus.PENDING
result: Optional[str] = None
error: Optional[str] = None
class TaskAgent:
def __init__(self, gpt_client, tool_executor):
self.gpt_client = gpt_client
self.tool_executor = tool_executor
self.current_task = None
def plan_task(self, user_request: str) -> List[TaskStep]:
# 让 GPT 分解任务
prompt = f"""
将以下用户请求分解为具体步骤，每个步骤要能明确执行或验证：
用户请求：{user_request}
返回格式（JSON）：
{{"steps": [{{"id": "1", "description": "步骤描述"}}, ...]}}
"""
response = self.gpt_client.chat_completion([
{"role": "system", "content": "你是一个任务规划专家。"},
{"role": "user", "content": prompt}
])
steps_data = json.loads(response)
return [
TaskStep(id=step["id"], description=step["description"])
for step in steps_data["steps"]
]
def execute_step(self, step: TaskStep, context: dict) -> str:
# 执行单个步骤
prompt = f"""
执行以下任务步骤：
步骤描述：{step.description}
上下文信息：{json.dumps(context, ensure_ascii=False)}
如果需要调用工具，使用提供的工具；如果信息不足，明确说明需要什么信息。
"""
response = self.gpt_client.chat_completion([
{"role": "system", "content": "你是一个任务执行专家。"},
{"role": "user", "content": prompt}
], tools=self.tool_executor.to_openai_format())
return response
def run_task(self, user_request: str) -> dict:
# 规划任务
steps = self.plan_task(user_request)
context = {"user_request": user_request}
results = []
for step in steps:
step.status = TaskStatus.IN_PROGRESS
try:
result = self.execute_step(step, context)
step.status = TaskStatus.COMPLETED
step.result = result
context[f"step_{step.id}_result"] = result
except Exception as e:
step.status = TaskStatus.FAILED
step.error = str(e)
# 简化处理：遇到错误就停止
break
results.append(step)
return {
"request": user_request,
"steps": results,
"status": "completed" if all(s.status == TaskStatus.COMPLETED for s in results) else "failed"
}
这个框架很粗糙，实际项目里会更复杂，但基本思路是一样的：把复杂任务分解成步骤，逐个执行，每个步骤的状态和结果都被记录下来。
几个踩过的坑
最后说几个印象比较深的坑：
Token 计算问题：tiktoken 的计算和 OpenAI 实际计费不完全一致，最好留一些余量。另外，工具调用的参数和返回值也算 Token，这个容易忽略。
并发安全问题：多用户同时调用同一个 GPT 客户端实例，可能会有问题。后来改成每个请求独立实例，或者用连接池。
模型升级陷阱：从 gpt-3.5 升级到 gpt-4 后，某些 Prompt 效果变差了，需要重新调优。模型不是越新越好，要看具体场景。
成本控制：早期没做成本统计，测试用户刷了几百次对话才发现超预算。后来加上每次调用的 Token 数和成本记录。
响应时间：gpt-4 的响应时间比 gpt-3.5 慢很多，用户体验明显下降。对于实时性要求高的场景，可能需要降级或缓存。
收尾
从一行 chat() 脚本到能调工具、管记忆、做任务分解，变化最大的认知是：让 AI 在真实场景里稳定干活，比学会调 API 难得多。
简单场景现在基本能 hold 住；复杂任务、长链路、多用户并发，边界还是一堆。技术迭代快，很多"最佳实践"隔半年就得重审。
我的做法一直是从小场景开跑，用真实用户反馈驱动迭代。有些问题不是 Prompt 写得差，就是当前模型能力到那了。第一次写的聊天脚本现在看很简陋，但后面所有改动都从那几行代码长出来的。
版权声明: 本文首发于 指尖魔法屋-关于GPT应用的几点记录（https://blog.thinkmoon.cn/post/167-gpt-app-deep-dive-from-chatbot-to-intelligent-assistant/） 转载或引用必须申明原指尖魔法屋来源及源地址！
评论
使用 GitHub 账号登录后即可留言，支持 Markdown。