关于AI系统架构的几点记录
按照传统思维，大致就是这么几个组件：
- 文档处理模块：负责把各种格式的文档转换成文本切片
- 向量数据库：存储文档的向量表示
- LLM 调用层：负责理解和生成
- 业务逻辑层：处理用户请求、路由、鉴权等
看架构图也很漂亮：
一开始的设想其实挺简单
当时的场景是做一个企业知识库问答系统，需求听起来不难：用户提问，系统从文档里找答案，用 LLM 组织语言返回。按照传统思维，大致就是这么几个组件：
- 文档处理模块：负责把各种格式的文档转换成文本切片
- 向量数据库：存储文档的向量表示
- LLM 调用层：负责理解和生成
- 业务逻辑层：处理用户请求、路由、鉴权等
看架构图也很漂亮：
图很清楚，逻辑也顺。但第一次真正跑起来就发现了问题：这个架构把每个组件都当成了"可靠的服务调用"，实际上 AI 组件压根不按这个剧本演。
第一个坑：把 LLM 当成普通 HTTP 服务
一开始的代码是这么写的（Python 为例）：
import httpx
async def query_llm(prompt: str) -> str:
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"model": "gpt-4",
"messages": [{"role": "user", "content": prompt}],
"timeout": 30
}
)
return response.json()["choices"][0]["message"]["content"]
这个函数在测试环境跑了几天都挺好，直到某天早上并发量突然翻倍。系统开始疯狂报错，大部分都是 Timeout 和 Connection 错误。
问题很明显：我把 LLM 调用当成了普通的 HTTP 请求，但实际上 LLM 服务有几个很明显的特点：
- 延迟不稳定：同样的请求，有时 500ms 返回，有时要 5 秒
- 并发限制严格：大多数提供商都有严格的 rate limit
- 成本随用量线性增长：失败重试要真金白银
当时的修复方案是加了一层调用管理：
class LLMCaller:
def __init__(self, max_concurrent=5, timeout=30):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.timeout = timeout
async def call(self, prompt: str, retry=3) -> str:
for attempt in range(retry):
async with self.semaphore:
try:
async with asyncio.timeout(self.timeout):
# 实际调用逻辑
result = await self._do_call(prompt)
return result
except asyncio.TimeoutError:
if attempt == retry - 1:
raise
await asyncio.sleep(2 ** attempt) # 指数退避
except httpx.HTTPStatusError as e:
if e.response.status_code == 429: # Rate limit
wait_time = int(e.response.headers.get("Retry-After", 60))
await asyncio.sleep(wait_time)
continue
raise
加了这层之后，稳定性确实好多了，但让我意识到一个更深层的问题：AI 组件的集成不能按传统微服务那套来做，必须要考虑其独特的不确定性。
第二个坑：向量检索和 LLM 的边界问题
一开始的设计是：先做向量检索，把 top-k 个文档片段喂给 LLM，让它组织答案。听起来很合理。
但实际跑起来发现几个问题：
- 检索质量参差不齐：有时候检索结果很差，LLM 拿到的是一堆不相关的文档
- 上下文长度爆炸：用户问题很简单，但为了确保检索充分，每次都塞了 10+ 个文档片段
- 成本不可控：每次请求都要跑一次 LLM，不管检索结果如何
当时的解决方案是加了一个"检索质量评估"组件：
async def enhanced_query(query: str) -> str:
# 先做快速检索
docs = await vector_search(query, top_k=3)
# 评估检索质量
relevance = await assess_relevance(query, docs)
if relevance < 0.5:
# 质量不够，扩展检索
docs = await vector_search(query, top_k=10)
relevance = await assess_relevance(query, docs)
if relevance < 0.3:
# 还是差，直接走通用回答
return await generic_llm_response(query)
# 质量可以，用 LLM 组织答案
return await llm_summarize(query, docs)
这个方案多了一层判断逻辑，但整体效果明显好转，而且成本也降了约 30%。让我认识到：AI 系统里的组件之间不是简单的串联，而是需要有"中间层"来做质量评估和路由决策。
第三个坑：状态管理和会话一致性
用户开始反馈"为什么问同样的问题，答案每次都不一样"。这是 AI 系统常见的问题：每次请求都是独立的，没有状态管理。
一开始我想了个简单的方案：把每次对话历史存到 Redis 里，下次请求时把历史一起塞给 LLM。代码大概是这样：
async def chat_with_history(user_id: str, message: str) -> str:
history = await redis.get(f"chat:{user_id}")
if history:
messages = json.loads(history)
else:
messages = []
messages.append({"role": "user", "content": message})
response = await llm_call(messages)
messages.append({"role": "assistant", "content": response})
await redis.set(f"chat:{user_id}", json.dumps(messages))
return response
这个方案跑了几天就崩了：Redis 内存爆炸，而且 LLM 的上下文长度限制很快就到了。
实际的修复方案是做了两层优化：
- 对话摘要：每过几轮对话，就把前面的历史用 LLM 总结一下，减少上下文长度
- 分级存储：最近几轮放 Redis，历史对话存到数据库，只保留关键信息
async def chat_with_memory(user_id: str, message: str) -> str:
# 获取最近对话
recent_history = await redis.get(f"chat:recent:{user_id}")
messages = json.loads(recent_history) if recent_history else []
# 检查是否需要摘要
if len(messages) > 10:
summary = await llm_summarize_history(messages[:5])
messages = messages[5:]
messages.insert(0, {"role": "system", "content": f"对话摘要：{summary}"})
messages.append({"role": "user", "content": message})
response = await llm_call(messages)
messages.append({"role": "assistant", "content": response})
# 只保留最近对话
await redis.set(f"chat:recent:{user_id}", json.dumps(messages[-10:]))
return response
这个方案虽然复杂了一点，但至少解决了内存和长度限制的问题。不过也让我意识到：AI 系统的状态管理比传统系统复杂，需要考虑模型的上下文限制和成本。
调整后的架构
经过这些折腾，调整后的架构图变成了这样：
这个架构虽然看起来复杂了一点，但每一层都是因为实际踩过的坑才加上去的。而且每个组件的职责都比较清晰：
- 调用管理器：处理 LLM 调用的并发、重试、超时
- 会话管理器：处理对话历史、摘要、分级存储
- 质量评估：在检索和 LLM 之间做质量判断
- 并发控制：控制 LLM 调用的并发数
一些经验总结
这次实践中积累的几个判断，可能会对后面做 AI 系统的人有点用：
- 不要把 AI 组件当成可靠服务：它们的延迟、成功率都和传统服务不一样，要做专门的适配层
- 组件之间需要"质量评估层"：特别是在检索和生成之间，不是简单的串联就行
- 成本控制要内置到架构里：LLM 调用是按次付费的，失败重试要考虑成本
- 状态管理要考虑上下文限制：不是所有历史都能塞给模型，要做摘要和分级存储
- 监控和日志很重要：AI 组件的不确定性，导致问题定位比传统系统难，需要有详细的调用日志
还在持续改进的地方
目前的架构虽然能用了，但还有一些问题没完全解决：
- 模型切换：不同场景用不同模型（有些用 GPT-4，有些用 GPT-3.5），现在还是硬编码
- 成本预警：用量突增时缺少预警机制，成本控制不够精细
- 缓存策略：相似问题的缓存做得还不够，浪费了不少调用量
- A/B 测试：新的组件或策略上线前，缺少自动化测试流程
这些都在逐步改进中，但每一项都不是简单的技术问题，需要结合业务和成本来权衡。
最后说一点感受
做 AI 系统架构和传统系统最大的不同，可能就是你不能假设所有组件都是"听话"的。它们有自己的脾气，有自己的限制，还有自己的成本模型。
传统系统设计时，你关注的是"如何让组件之间高效协作"；AI 系统设计时，你还要考虑"如何让不确定的组件协作得尽量稳定"。
这不是一个简单的技术问题，更多是对系统复杂性的重新理解。可能也是为什么很多 AI 系统，初版看起来简单，但真正要落地到生产环境，反而要花更多时间在架构优化上。
不过好在，坑踩多了，总能总结出一些可以复用的模式。希望这篇文章的经验，能帮后来者少走一点弯路。
参考：
- OpenAI API 官方文档：https://platform.openai.com/docs/
- LangChain 文档：https://python.langchain.com/
- 向量数据库对比：https://www.pinecone.io/learn/vector-database/
版权声明: 本文首发于 指尖魔法屋-关于AI系统架构的几点记录（https://blog.thinkmoon.cn/post/217-ai-system-architecture-components-integration/） 转载或引用必须申明原指尖魔法屋来源及源地址！
评论
使用 GitHub 账号登录后即可留言，支持 Markdown。