写在前面的话
前行之路。昨天，侄子给我打电话，聊了会天。侄子今年高三，我感觉他很礼貌，说话也有方式，我上高三时，根本说不出来这么有水平的话。侄子说了下规划，先把英语拿下，3月份考完，成绩满意，再把英语放下，再攻其它科，思路很清晰。我当时就是跟着老师的计划走，侄子有自己的计划，比较主动。加上侄子属于新疆内高班特招计划，有这个优势，会好很多。哥说，上个一本没问题，但985，211有点费劲，侄子去年说上985是没问题的，看来还是过于自信了。也可能是在天津读书这三年，有一段时间玩游戏，耽误了学习，但这也是没办法的事，就像我当年，明明知道考上好大学，一切光明，但上课就是胡思乱想，影响了学习。一个人在前行的路上时，总会遇到阻碍，有些阻碍，父母干着急帮不上忙，只能靠自己，祝愿侄子超常发挥，考上好大学吧。
关键词：Python、chat_sev、同步聊天记录、分科处理。
一、chat_serve相关功能开发
1. 同步新的聊天记录
需求描述：当用户进行新的聊天时，需要实时地将聊天记录同步到 chat_sev 服务中的 conversation 数据表里。我们先来看看具体的实现代码。
开工。
第一步：查看与修改接口代码
a. 审视现有聊天生成代码
20241216时间段：1532-1551
当前想到一个方法：在新的聊天内容生成完毕后，立即更新对应的消息记录。
b. 关于条件更新的思考
20241216时间段：1551-1600
条件更新的逻辑暂时不做，我们目前先实现一个增量返回的机制。核心思路是，在流式返回结束时，触发一次同步保存。前端控制器中的 WebSocket 端点代码修改如下：
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
data = await websocket.receive_text()
req = json.loads(data)
response = chat(req['conversation_id'],req['question'])
res_str = ''
for msg in response:
if res_str:
res_str = res_str + msg
else:
res_str = msg
if msg.endswith(b'\n\n'):
json_str = res_str.decode('utf-8')
if 'data:{"retcode": 666' in json_str:
sync_memssages_rag(req['conversation_id'])
break
else:
await websocket.send_json(json_str)
res_str = ''
except WebSocketDisconnect:
print("Client disconnected")
注：服务端在流式传输结束时，会发送一个特定的结束标志（这里用了 retcode: 666）。前端控制器在检测到这个标志后，调用 sync_memssages_rag 函数执行同步。
后端的同步函数 sync_memssages_rag 实现如下：
def sync_memssages_rag(conversation_id:str) ->str:
"""
生成新的会话，返回会话id
"""
data = get_conversation(conversation_id)
log.info(f”data {data}”)
log.info(f”data-message {data[‘message’]}”)
if data :
ConversationService.update_by_id(conversation_id, {“message”:json.dumps(data[‘message’],ensure_ascii=False)})
return ‘success’
这个函数根据 conversation_id 获取最新的对话数据，然后更新到数据库对应的记录中。
c. 代码部署上线
20241216时间段：1837-1840
首先部署更新后的 chat_sev 服务。上线过程顺利，没有出现问题。接下来，可以考虑优化流式返回的体验，比如分段返回。
d. 关于分段返回的考量
20241216时间段：1900-1920
分段返回的功能暂时搁置。经过分析，当前响应速度的瓶颈并非中间返回延迟，而是首次生成答案的速度。所以，我们优先处理其他更影响体验的功能。接下来，开始实现“分科处理”功能。
第二步：功能测试
对同步聊天记录的功能进行了测试，效果符合预期。
2. 分科处理（基于用户意图识别）
需求描述：需要根据用户提供的病情描述，或者引导用户选择一个咨询科室。这本质是一个用户意图识别的任务。
分析：同事“龙哥”已经实现了一部分代码，我们可以参考他的思路。
开工。
第一步：查看现有代码与设计思路
阅读代码后，我的思路是：在数据库的会话表中增加相关字段。例如，标记“是否已进行意图识别”。如果未识别，则在回答用户问题前先执行意图识别。
首先，我们需要定义用户可能的意图类别。参考已有的设计，类别包括：疾病问诊、疫苗接种、用药咨询、在线复诊、驱虫选药、科学养宠、行为异常、检测报告。
设计要点：
- 根据用户的提问内容进行意图识别。
- 如果识别不够准确，可以准备几个内置的引导性问题进行澄清。
- 在数据库增加两个字段：
is_intent (是否已识别) 和 intention (识别结果)。
- 后期可能需要考虑“意图重识别”，因为用户在对话中可能会切换话题。但前期我们先实现一次性的初始意图识别。
第二步：为数据库表添加字段
20241217周二时间段：1124-1139
执行如下 SQL 语句，为 conversation 表添加字段：
ALTER TABLE `rag_flow`.`api_4_conversation`
ADD COLUMN `is_intent` tinyint(1) NULL DEFAULT 0 COMMENT ‘是否对用户意图进行识别 0没有 1已识别’ AFTER `name`,
ADD COLUMN `intention` varchar(50) NULL COMMENT ‘对用户进行意图识别的内容’ AFTER `is_intent`
字段添加成功后，下一步就是在用户提问时，调用意图识别服务，并将结果写入这些字段。
第三步：实现意图识别逻辑
a. 定位提示词与核心函数
20241217周二时间段：1139-1200
我们需要在回答用户问题之前，插入意图识别的步骤。先找到已实现的识别函数。龙哥编写的核心函数如下：
def user_intension_congnition(conversation_id:str,question:str):
"""
用户意图识别
返回：以下分类的一种
疾病问诊
疫苗接种
用药咨询
在线复诊
驱虫选药
科学养宠
行为异常
检测报告
"""
question = user_intension_reconition_prompt.format(input_text=question)
return chat(conversation_id, question,stream=False)
这个函数接收用户问题，并将其套入一个精心设计的提示词模板中，然后调用大模型 chat 函数（非流式）来获取分类结果。其中，user_intension_reconition_prompt 是定义好的提示词模板字符串，内容如下：
user_intension_reconition_prompt = ””
你是一个宠物医生，擅长对用户的问题进行有效分类，分类内容必须是以下分类的中的一种：
疾病问诊
疫苗接种
用药咨询
在线复诊
驱虫选药
科学养宠
行为异常
检测报告
######################
-Examples-
######################
Example 1:
Question:
你好，医生。我的狗狗最近总是打喷嚏，特别是早上和晚上，偶尔还会流鼻涕。家里没有太多灰尘，也没有发现其他异常。我想知道这可能是过敏还是别的问题？我该怎么处理？
################
Output:
疾病问诊
#############################
Example 2:
你好，医生。我家的狗狗最近好像感冒了，流鼻涕、有点咳嗽，精神状态也不太好。我家里有一些人类用的感冒药，请问可以给它吃吗？如果可以，剂量怎么控制？
################
Output:
用药咨询
#############################
Example 3:
你好，医生。您帮我看下我家狗狗的血常规检测报告，是否感染流感了？
################
Output:
检测报告
#############################
Example 4:
你好，医生。我家猫咪最近总是频繁上厕所，但每次尿量都很少，有时甚至没有尿出来。我很担心，这是不是泌尿系统的问题？请问该如何治疗？
################
Output:
行为异常
#############################
-Real Data-
######################
Question:{input_text}
######################
Output:
“”
这个提示词模板通过 few-shot 示例，明确地引导大模型输出我们预设的类别之一。
b. 编写测试接口进行验证
20241217周二时间段：1213-1220
为了验证意图识别功能是否正常工作，我编写了一个简单的 HTTP 接口进行测试。
首先引入相关模块：
from service.pet_doctor import user_intension_congnition
然后编写接口路由：
######测试意图识别
######20241217
@router.post(‘/intention’)
async def intention(request:Request):
req = await request.body()
req = json.loads(req.decode(“utf-8”))
conversation_id = ”
if ‘conversation_id’ in req:
conversation_id = req[‘conversation_id’]
if not conversation_id:
return error(‘1001’,’conversation_id-是必传参数’)
question = ”
if ‘question’ in req:
question = req[‘question’]
if not conversation_id:
return error(‘1002’,’question-是必传参数’)
response = user_intension_congnition(conversation_id,question)
log.info(f”response{get_caller_line_number()} {response}”)
res_str = ”
for msg in response:
log.info(f”msg{get_caller_line_number()} {msg}”)
if res_str:
res_str = res_str + msg
else:
res_str = msg
if msg.endswith(b’\n\n’):
json_str = res_str.decode(‘utf-8’)
log.info(f”json_str{get_caller_line_number()} {json_str}”)
if ‘data:{”retcode”: 666’ in json_str:
# sync_memssages_rag(req[‘conversation_id’])
break
else:
res_str = ”
return ‘dddd’
然而，在测试这个接口时，程序报错了，返回了“Not Found”等信息。下一步就需要排查这个错误的原因，是路径问题、服务状态问题还是函数逻辑本身有缺陷。这部分调试工作，就留到下次再继续了。
二、生活掠影
拍摄于2025年6月1日，13:15:50，带二宝在小区里玩，二宝当时两岁八个月。很多时候，自己心里很明白，有些事这样做是错的，但就是没办法。我读高三那会，明明知道自己坐教室里啥也不干都冒汗，想到父母在地里除草，给羊割草，很辛苦，自己应该好好学习，但老师讲课，我就不自觉的想父母的辛苦，自己的畜生，常进入幻境，下了课，找个没人的地方，卡卡扇自己嘴巴子，下节课还是走神。就像侄子，明明知道自己有这个内高班的优势，也知道自己无比聪明，也能理解父母的辛苦，可能游戏太有诱惑力了，实在控制不住自己，打完游戏我相信他也后悔，但自己不受控制。不过，好在侄子已经意识到自己落后了，接下来的四个多月准备拼一把，大力出奇迹，相信一定会成功的，加油！
三、昨日花销
记录一下日常开销，也是让自己对消费更有概念。
《本文完》
以上就是本次关于 Python 后端服务开发中，处理聊天记录同步和用户意图识别的一些实践记录。整个过程涉及到 WebSocket 通信、数据库 (MySQL) 操作以及与大模型 (AI) 的提示工程交互，是一次挺典型的全栈小功能开发。如果你对这类开发者日常的实践和踩坑记录感兴趣，欢迎来 云栈社区 的开发者广场逛逛，那里有很多类似的干货分享和技术吐槽。
发表于 2026-2-11 20:06:25
|
查看: 28|
回复: 0
