boss_api 后端 LLM 代码学习笔记最近把项目里llm和app/apis/llm下面的代码过了一遍从最简单的单轮调用到多轮对话、Redis 存消息、流式输出、历史压缩一点一点搭起来的。写个笔记记下来省得以后忘了。1. 最简单的一问一答case1.py这文件最基础就是调用一下通义千问。先整一个OpenAI客户端把api_key和base_url填进去。这里的base_url是阿里百炼的兼容地址api_key从环境变量DASHSCOPE_API_KEY读。模型用的qwen-plus。clientOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://输入自己的api.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)然后发消息结构是messages[system消息, user消息]system用来告诉模型它是个智能助手。调完接口从completion.choices[0].message.content里取回答。这种写法一次只问一句模型也不会记得你上一句说了啥。2. 流式输出case2.pycase1.py有个问题要等模型把整段话想完才一次性吐出来。如果回答很长前端就干等着体验不太好。流式就是改一行streamTrue。模型想出一个字就给你一个字你可以边收边展示。completionclient.chat.completions.create(modelqwen-plus,messages[...],streamTrue,stream_options{include_usage:True})返回的completion是个可迭代对象每次循环拿到一个chunk。chunk.choices[0].delta.content就是这一小段增量文本。把它们累加起来就是完整回答。3. 多轮对话case3.py单轮对话只能你问一句、它答一句再问新问题时模型不知道上下文。多轮对话就是维护一个messages列表每次把用户问题和模型回答都塞进去下一次请求时把整个列表再传回去。messages[]messages.append({role:user,content:推荐一部科幻电影})assistant_outputget_response(messages)messages.append({role:assistant,content:assistant_output})messages.append({role:user,content:这部电影导演是谁})assistant_outputget_response(messages)这样第二轮问导演是谁的时候模型看到前面有推荐科幻电影那段就能知道这部电影指的是啥。4. 用 Redis 把对话存下来case4.pycase3.py的messages存在内存里程序一关就没了。真实的聊天系统肯定不能这样得持久化。这里用了 Redis 的list。key比如叫boss:llm:case4:message:3后面的3可以理解为用户 ID 或者某个标识。每次新消息用rpush追加到列表末尾想读历史就用lrange(key, 0, -1)全取出来。redis_client.rpush(key,json.dumps(message1,ensure_asciiFalse))redis_messagesredis_client.lrange(key,0,-1)messages[json.loads(message)formessageinredis_messages]Redis 里存的是 JSON 字符串所以写入时要dumps读取时要loads。5. 把脚本包成接口case1_api.py前面都是本地脚本项目里要用 HTTP 接口暴露出去所以写了case1_api.py用 FastAPI 的APIRouter注册两个路由。/llm-day01/case1是普通接口等模型返回完整结果再一次性返回给前端completionclient.chat.completions.create(...)ai_replycompletion.choices[0].message.contentreturn{code:1,data:ai_reply}/llm-day01/case2是流式接口返回StreamingResponsemedia_type 是text/event-stream。stream_chunk是个生成器每收到一段模型输出就yield出去前端就能实时看到文字出现。yieldfdata:{delta.content}\n\nyielddata:[done]\n\n这里data:是 SSEServer-Sent Events的格式前端按这个规范解析就行。6. 完整的多轮对话系统case2_api.py这个文件是重头戏里面实现了四个接口也是前端 AI 助手页面真正调用的那套。6.1 创建会话/create_session用户点新对话时调用。给当前用户生成一个session_id用uuid.uuid4()标题默认叫新会话再记录创建时间。然后塞进 Rediskeyfboss:llm:session:{user_id}redis_client.rpush(key,json.dumps(current_session,ensure_asciiFalse))一个用户的所有会话都存在这个 key 对应的列表里。6.2 查询会话列表/get_session_list就是把上面那个列表读出来转成字典数组返回。前端左侧的历史会话就是这么渲染的。6.3 发送消息/send_message这是最复杂的一个。流程大概这样先拼出这个会话的消息 keyboss:llm:case2:messages:{user_id}:{session_id}。查 Redis如果这个会话一条消息都没有说明是第一次聊先塞一条 system 消息进去告诉模型你是个智能助手。同时把会话标题改成用户首条消息的前 10 个字。把用户的新消息rpush进去带上发送时间。把 Redis 里的全部消息读出来做压缩下面再说。调stream_ai_response流式返回模型回答。等流走完之后再把完整回答写回 Redis。为什么 AI 回复要等流完再写 Redis因为流式过程中只有片段最后拼成完整内容才能存。下次进这个会话查历史看到的就是完整的assistant消息。6.4 查询会话详情/get_user_session_messages前端点左侧某个历史会话时调用把user_id和session_id一拼从 Redis 读出这个会话的所有消息返回给前端渲染聊天内容。7. 消息压缩是干嘛的多轮对话聊久了messages会越来越长。太长有两个问题一是 token 贵二是模型可能看不过来。所以写了个compression_messages函数。逻辑很简单如果消息没超过 10 条就不压缩。超过 10 条的话把最早的那部分丢给模型让它自己总结成一段摘要然后保留最近 10 条完整消息。最后返回的是[{role:user,content:摘要}]最近10条消息这样模型既能看到很久以前聊的核心内容又能看到最近的具体上下文。8. 请求参数校验schemas/llm_case1.pyFastAPI 用 Pydantic 做入参校验。LLMCase1只有一个question对应/llm-day01的两个接口。LLMCase2有user_id、session_id、message对应/llm-day02的发消息接口。classLLMCase2(BaseModel):user_id:intsession_id:strmessage:str这样接口收到的参数会先校验类型类型不对直接返回 422省得后面代码里还要自己判断。9. 路由注册main.py最后这些 router 要在main.py里include_router才会生效fromapp.apis.llm.case1_apiimportllm_day01_routerfromapp.apis.llm.case2_apiimportllm_day02_router app.include_router(llm_day01_router)app.include_router(llm_day02_router)前缀在 router 自己里面定义了llm_day01是/llm-day01llm_day02是/llm-day02。10. 一些踩过的坑环境变量要生效DASHSCOPE_API_KEY配了但程序读不到很多时候是 IDE 或者终端没重启环境变量没加载进来。Redis 得先启动llm_day02这几个接口都依赖 Redis没起 Redis 会报连接错误。流式被缓冲StreamingResponse默认可能被 Nginx 或者某些代理缓冲导致前端看着像一次性返回。后面加了几个 headerCache-Control: no-cache、X-Accel-Buffering: no、Connection: keep-alive。前端响应式问题前端用 Vue 接流式内容时如果直接改普通对象的字段界面不会实时刷新要用reactive或者改ref数组里的对象代理。大概就是这样。后面如果要加新功能比如上传文件让模型读、切换不同模型、或者给不同会话设置不同提示词估计也是在这几个接口上继续改。先记到这。