項目背景介紹

sanic-web

項目地址：https://github.com/apconw/sanic-web

一個輕量級、支持全鏈路且易于二次開發的大模型應用項目(Large Model Data Assistant) 支持DeepSeek/Qwen2.5等大模型 基于 Dify 、Ollama&Vllm、Sanic 和 Text2SQL 📊 等技術構建的一站式大模型應用開發項目，采用 Vue3、TypeScript 和 Vite 5 打造現代UI。它支持通過 ECharts 📈 實現基于大模型的數據圖形化問答，具備處理 CSV 文件 📂 表格問答的能力。同時，能方便對接第三方開源 RAG 系統 檢索系統 🌐等，以支持廣泛的通用知識問答。

這個項目可以作為text2sql的經典案例，通過自然語言來訪問業務數據庫，最終使用echarts圖表可視化展示分析數據。使用了獨立開發的web頁面，對于前端小伙伴來說也比較友好，這完全可以作為一個AI智能助手的Demo實現。

Dify_service handle_think_tag報錯NoneType

問題描述

debug

修改services/dify_service.py/handle_think_tag代碼，如下：

    @staticmethodasync def handle_think_tag(answer):"""處理<think>標簽內的內容:param answer""""""處理<think>標簽內的內容，或JSON格式的thoughts字段:param answer"""think_content = ""remaining_content = answer# 會遇到answer可能不能解析到，先嘗試解析為JSONtry:data = json.loads(answer)if isinstance(data, dict) and "thoughts" in data:think_content = data["thoughts"]remaining_content = answerreturn think_content, remaining_contentexcept Exception:pass# 再嘗試正則提取<think>標簽match = re.search(r"<think>(.*?)</think>", answer, re.DOTALL)if match:think_content = match.group(1)remaining_content = re.sub(r"<think>.*?</think>", "", answer, flags=re.DOTALL).strip()return think_content, remaining_content# 如果都沒有，返回空return "", answer

Dify調用不成功，一直轉圈圈

問題描述

debug

檢查本地dify的端口號，修改sanic-web/docker/docker-compose.yaml中dify端口號到本地端口號，比如原端口號是18000，修改成80。

  chat-service:image: apconw/sanic-web:1.1.2container_name: sanic-webenvironment:- ENV=test- DIFY_SERVER_URL=http://host.docker.internal:80- DIFY_DATABASE_QA_API_KEY=app-AXDUw8TtcY7N6TMGHkPaC4VF- MINIO_ENDPOINT=host.docker.internal:19000- MINIO_ACCESS_KEY=sIR5eeDkiwoo779yNJbw- MiNIO_SECRET_KEY=MreuQ3aC1ymHJeo3QfzSg7aPz7PqlxeOw39nZUdEports:- "8088:8088"extra_hosts:- "host.docker.internal:host-gateway"

前端markdown格式只顯示前5頁

問題描述

debug

1. 修改代碼

修改web/src/components/MarkdownPreview/MarkdownTable.vue第39行-40行代碼，將:data="pagedTableData"改為:data=“tableData”，并移除:pagination="pagination"屬性：

<template><div style="background-color: #ffffff"><n-cardtitle="表格"embeddedbordered:content-style="{ 'background-color': '#ffffff' }":header-style="{color: '#26244c',height: '10px','background-color': '#f0effe','text-align': 'left','font-size': '14px','font-family': 'PMingLiU'}":footer-style="{color: '#666','background-color': '#ffffff','text-align': 'left','font-size': '14px','font-family': 'PMingLiU'}"><divstyle="display: flex;justify-content: space-between;margin-bottom: 10px;"></div><n-data-tablestyle="height: 550px;width: 850px;margin: 0px 10px;background-color: #ffffff;":columns="columns":data="tableData":max-height="550"virtual-scrollvirtual-scroll-x:scroll-x="scrollX":min-row-height="minRowHeight":height-for-row="heightForRow"virtual-scroll-header:header-height="48"/><template #footer>數據來源: 大模型生成的數據, 以上信息僅供參考</template></n-card></div>
</template>

2.重新構建1.1.3鏡像

# 進入web的docker目錄
cd docker
# 查看原始Dockerfile ，這步也可以省略
cat Dockerfile
# 使用原始Dockerfile構建新鏡像 
docker build -t apconw/chat-vue3-mvp:1.1.3 -f Dockerfile ..

3.更新sanic-web/docker/docker-compose.yaml

services:chat-web:image: apconw/chat-vue3-mvp:1.1.3  # 更新為新版本container_name: chat-vue3-mvpports:- "8081:80"extra_hosts:- "host.docker.internal:host-gateway"depends_on:- chat-service

4. 重新部署

docker-compose down
docker-compose up -d

Dify超時60秒，服務器報錯

問題描述

2025/05/09 08:16:19 [error] 20#20: *1 upstream timed out (110: Operation timed out) while reading response header from upstream, client: 192.168.65.1, server: localhost, request: “POST /sanic/dify/get_answer HTTP/1.1”, upstream: “http://192.168.65.254:8088/dify/get_answer”, host: “localhost:8081”, referrer: “http://localhost:8081/chat”

---

192.168.65.1 - - [09/May/2025:08:16:19 +0000] “POST /sanic/dify/get_answer HTTP/1.1” 504 497 “http://localhost:8081/chat” “Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36” “-”

---

192.168.65.1 - - [09/May/2025:08:16:19 +0000] “POST /sanic/dify/get_dify_suggested HTTP/1.1” 200 63 “http://localhost:8081/chat” “Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36” “-”

debug

修改dify_service.py中的DiFyRequest，添加心跳機制：

1. 添加心跳機制:

• 新增send_heartbeat方法發送SSE心跳
• 在處理請求過程中每10秒發送一次心跳
• 在開始、結束和錯誤處理時也發送心跳

2. 增強錯誤處理:

• 添加send_error_message方法，統一錯誤信息發送
• 完善異常處理，確保錯誤信息能發送到客戶端

3. 增加超時時間:

• 將aiohttp的超時設置從2分鐘增加到5分鐘

4. 確保連接正確關閉:

• 在finally塊中確保發送最后的心跳消息
• 確保調用res_end方法正確關閉連接

class DiFyRequest:"""DiFy操作服務類"""def __init__(self):passasync def exec_query(self, res):"""執行查詢并處理流式響應"""try:# 獲取請求體內容 從res流對象獲取request-bodyreq_body_content = res.request.body# 將字節流解碼為字符串body_str = req_body_content.decode("utf-8")req_obj = json.loads(body_str)logging.info(f"query param: {body_str}")# str(uuid.uuid4())chat_id = req_obj.get("chat_id")qa_type = req_obj.get("qa_type")#  使用正則表達式移除所有空白字符（包括空格、制表符、換行符等）query = req_obj.get("query")cleaned_query = re.sub(r"\s+", "", query)# 獲取登錄用戶信息token = res.request.headers.get("Authorization")if not token:raise MyException(SysCodeEnum.c_401)if token.startswith("Bearer "):token = token.split(" ")[1]# 封裝問答上下文信息qa_context = QaContext(token, cleaned_query, chat_id)# 判斷請求類別app_key = self._get_authorization_token(qa_type)# 構建請求參數dify_service_url, body_params, headers = self._build_request(chat_id, cleaned_query, app_key, qa_type)# 收集流式輸出結果t02_answer_data = []# 收集業務數據流式輸出結果t04_answer_data = {}# 發送初始連接消息await self.send_heartbeat(res, "開始處理請求")# 心跳計時器last_heartbeat = time.time()async with aiohttp.ClientSession(read_bufsize=1024 * 16) as session:async with session.post(dify_service_url,headers=headers,json=body_params,timeout=aiohttp.ClientTimeout(total=60 * 5),  # 增加到5分鐘超時) as response:logging.info(f"dify response status: {response.status}")if response.status == 200:data_type = ""bus_data = ""while True:# 發送心跳保持連接current_time = time.time()if current_time - last_heartbeat > 10:  # 每10秒發送一次心跳await self.send_heartbeat(res, "處理中...")last_heartbeat = current_timereader = response.contentreader._high_water = 10 * 1024 * 1024  # 設置為10MBchunk = await reader.readline()if not chunk:# 發送最后的心跳await self.send_heartbeat(res, "讀取數據完成")breakstr_chunk = chunk.decode("utf-8")# 處理數據塊if str_chunk.startswith("data"):# 更新最后心跳時間last_heartbeat = time.time()str_data = str_chunk[5:]data_json = json.loads(str_data)event_name = data_json.get("event")conversation_id = data_json.get("conversation_id")message_id = data_json.get("message_id")task_id = data_json.get("task_id")# 處理消息事件...# 這里保留原有的事件處理邏輯if DiFyCodeEnum.MESSAGE.value[0] == event_name:answer = data_json.get("answer")if answer and answer.startswith("dify_"):event_list = answer.split("_")if event_list[1] == "0":# 輸出開始data_type = event_list[2]if data_type == DataTypeEnum.ANSWER.value[0]:await self.send_message(res,answer,{"data": {"messageType": "begin"}, "dataType": data_type},)elif event_list[1] == "1":# 輸出結束data_type = event_list[2]if data_type == DataTypeEnum.ANSWER.value[0]:await self.send_message(res,answer,{"data": {"messageType": "end"}, "dataType": data_type},)# 輸出業務數據elif bus_data and data_type == DataTypeEnum.BUS_DATA.value[0]:res_data = process(json.loads(bus_data)["data"])await self.send_message(res,answer,{"data": res_data, "dataType": data_type},)t04_answer_data = {"data": res_data, "dataType": data_type}data_type = ""elif len(data_type) > 0:# 這里輸出 t02之間的內容if data_type == DataTypeEnum.ANSWER.value[0]:await self.send_message(res,answer,{"data": {"messageType": "continue", "content": answer}, "dataType": data_type},)t02_answer_data.append(answer)# 這里設置業務數據if data_type == DataTypeEnum.BUS_DATA.value[0]:bus_data = answerelif DiFyCodeEnum.MESSAGE_ERROR.value[0] == event_name:# 輸出異常情況日志error_msg = data_json.get("message")logging.error(f"Error 調用dify失敗錯誤信息: {data_json}")await res.write("data:"+ json.dumps({"data": {"messageType": "error", "content": "調用失敗請查看dify日志,錯誤信息: " + error_msg},"dataType": DataTypeEnum.ANSWER.value[0],},ensure_ascii=False,)+ "\n\n")elif DiFyCodeEnum.MESSAGE_END.value[0] == event_name:t02_message_json = {"data": {"messageType": "continue", "content": "".join(t02_answer_data)},"dataType": DataTypeEnum.ANSWER.value[0],}print(t02_message_json)if t02_message_json:await self._save_message(t02_message_json, qa_context, conversation_id, message_id, task_id, qa_type)if t04_answer_data:await self._save_message(t04_answer_data, qa_context, conversation_id, message_id, task_id, qa_type)t02_answer_data = []t04_answer_data = {}except Exception as e:logging.error(f"Error during get_answer: {e}")traceback.print_exception(e)# 發送錯誤信息await self.send_error_message(res, f"處理請求出錯: {str(e)}")return {"error": str(e)}  # 返回錯誤信息作為字典finally:# 確保連接正確關閉await self.send_heartbeat(res, "請求處理完成")await self.res_end(res)async def send_heartbeat(self, res, message="心跳"):"""發送心跳信息保持連接活躍"""try:await res.write(f"data:{json.dumps({'heartbeat': True, 'message': message}, ensure_ascii=False)}\n\n")except Exception as e:logging.error(f"發送心跳失敗: {e}")async def send_error_message(self, res, error_message):"""發送錯誤信息"""try:await res.write("data:"+ json.dumps({"data": {"messageType": "error", "content": error_message},"dataType": DataTypeEnum.ANSWER.value[0],},ensure_ascii=False,)+ "\n\n")except Exception as e:logging.error(f"發送錯誤信息失敗: {e}")@staticmethodasync def handle_think_tag(answer):"""處理<think>標簽內的內容:param answer""""""處理<think>標簽內的內容，或JSON格式的thoughts字段:param answer"""think_content = ""remaining_content = answer# 會遇到answer可能不能解析到，先嘗試解析為JSONtry:data = json.loads(answer)if isinstance(data, dict) and "thoughts" in data:think_content = data["thoughts"]remaining_content = answerreturn think_content, remaining_contentexcept Exception:pass# 再嘗試正則提取<think>標簽match = re.search(r"<think>(.*?)</think>", answer, re.DOTALL)if match:think_content = match.group(1)remaining_content = re.sub(r"<think>.*?</think>", "", answer, flags=re.DOTALL).strip()return think_content, remaining_content# 如果都沒有，返回空return "", answer@staticmethodasync def _save_message(message, qa_context, conversation_id, message_id, task_id, qa_type):"""保存消息記錄并發送SSE數據:param message::param qa_context::param conversation_id::param message_id::param task_id::param qa_type::return:"""# 保存用戶問答記錄 1.保存用戶問題 2.保存用戶答案 t02 和 t04if "content" in message["data"]:await add_question_record(qa_context.token, conversation_id, message_id, task_id, qa_context.chat_id, qa_context.question, message, "", qa_type)elif message["dataType"] == DataTypeEnum.BUS_DATA.value[0]:await add_question_record(qa_context.token, conversation_id, message_id, task_id, qa_context.chat_id, qa_context.question, "", message, qa_type)async def send_message(self, response, answer, message):"""SSE 格式發送數據，每一行以 data: 開頭"""if answer.lstrip().startswith("<think>"):# 處理deepseek模型思考過程樣式think_content, remaining_content = await self.handle_think_tag(answer)# 發送<think>標簽內的內容message = {"data": {"messageType": "continue", "content": "> " + think_content.replace("\n", "") + "\n\n" + remaining_content},"dataType": "t02",}await response.write("data:" + json.dumps(message, ensure_ascii=False) + "\n\n")else:await response.write("data:" + json.dumps(message, ensure_ascii=False) + "\n\n")@staticmethodasync def res_begin(res, chat_id):""":param res::param chat_id::return:"""await res.write("data:"+ json.dumps({"data": {"id": chat_id},"dataType": DataTypeEnum.TASK_ID.value[0],})+ "\n\n")@staticmethodasync def res_end(res):""":param res::return:"""await res.write("data:"+ json.dumps({"data": "DONE","dataType": DataTypeEnum.STREAM_END.value[0],})+ "\n\n")@staticmethoddef _build_request(chat_id, query, app_key, qa_type):"""構建請求參數:param chat_id: 對話id:param app_key: api key:param query: 用戶問題:param qa_type: 問答類型:return:"""# 通用問答時，使用上次會話id 實現多輪對話效果conversation_id = ""if qa_type == DiFyAppEnum.COMMON_QA.value[0]:qa_record = query_user_qa_record(chat_id)if qa_record and len(qa_record) > 0:conversation_id = qa_record[0]["conversation_id"]body_params = {"query": query,"inputs": {"qa_type": qa_type},"response_mode": "streaming","conversation_id": conversation_id,"user": "abc-123",}headers = {"Content-Type": "application/json","Authorization": f"Bearer {app_key}",}dify_service_url = DiFyRestApi.build_url(DiFyRestApi.DIFY_REST_CHAT)return dify_service_url, body_params, headers@staticmethoddef _get_authorization_token(qa_type: str):"""根據請求類別獲取api/token固定走一個dify流app-IzudxfuN8uO2bvuCpUHpWhvH master分支默認的數據問答key:param qa_type:return:"""# 遍歷枚舉成員并檢查第一個元素是否與測試字符串匹配for member in DiFyAppEnum:if member.value[0] == qa_type:return os.getenv("DIFY_DATABASE_QA_API_KEY")else:raise ValueError(f"問答類型 '{qa_type}' 不支持")