|
|
@@ -4,7 +4,6 @@ from fastapi import APIRouter , UploadFile
|
|
|
from pydantic import BaseModel,Field
|
|
|
from requests import request
|
|
|
from agent.core.sign_check import check
|
|
|
-import logging
|
|
|
import agent.core.dao as dao
|
|
|
from asyncer import asyncify
|
|
|
from agent.core.vector import get_embeddings
|
|
|
@@ -15,6 +14,7 @@ import uuid
|
|
|
from fastapi import BackgroundTasks
|
|
|
from typing import Optional
|
|
|
import json
|
|
|
+from agent.logger import logger
|
|
|
|
|
|
router = APIRouter(prefix="/v1", tags=["AI Tagging"])
|
|
|
|
|
|
@@ -47,7 +47,6 @@ async def execute_reg(log_id:str,tag_category_id:str,phrase: str)-> list:
|
|
|
try:
|
|
|
for label in labels:
|
|
|
reg = label[1]
|
|
|
- logging.info(f"Executing regex for label_id {label[0]}")
|
|
|
if reg is not None:
|
|
|
pattern = re.compile(reg, re.VERBOSE)
|
|
|
if pattern.match(phrase):
|
|
|
@@ -55,13 +54,13 @@ async def execute_reg(log_id:str,tag_category_id:str,phrase: str)-> list:
|
|
|
else:
|
|
|
result.append(label[0])
|
|
|
except Exception as e:
|
|
|
- logging.error(f"Regex execution failed: {e}")
|
|
|
- logging.info(f"Regex filtering candidates: {result}")
|
|
|
+ logger.error(f"Regex execution failed: {e}")
|
|
|
+ logger.info(f"Regex filtering candidates: {result}")
|
|
|
dao.execute(
|
|
|
"""UPDATE ai_tagging.aitag_tag_log SET reg_result = %s WHERE id = %s""",
|
|
|
(str(result), log_id)
|
|
|
)
|
|
|
- logging.info(f"Updated reg_result for log_id {id}")
|
|
|
+ logger.info(f"Updated reg_result for log_id {id}")
|
|
|
return result
|
|
|
|
|
|
def vector_similarity_search(phrase: str, ids:list)-> list:
|
|
|
@@ -69,7 +68,7 @@ def vector_similarity_search(phrase: str, ids:list)-> list:
|
|
|
query = get_embeddings([phrase])[0]
|
|
|
results = hybrid_search(ids, query, top_k=TOP_K)
|
|
|
# return [{"id": r["_id"], "score": r["_score"], "tag_prompt": r["_source"]["tag_prompt"],"tag_name": r["_source"]["tag_name"],"tag_code": r["_source"]["tag_code"]} for r in results]
|
|
|
- return [{"id": r["_id"], "tag_remark":r["_source"]["tag_remark"], "tag_prompt": r["_source"]["tag_prompt"],"tag_name": r["_source"]["tag_name"],"tag_code": r["_source"]["tag_code"],"tag_path": r["_source"]["tag_path"] } for r in results]
|
|
|
+ return [{"id": r["_id"], "tag_remark":r["_source"]["tag_remark"], "tag_prompt": r["_source"]["tag_prompt"],"tag_name": r["_source"]["tag_name"],"tag_code": r["_source"]["tag_code"],"tag_path": r["_source"]["tag_path"],"category_id": r["_source"]["category_id"] } for r in results]
|
|
|
|
|
|
def init_tag_log(request: TaggingRequest):
|
|
|
id = uuid.uuid4().hex
|
|
|
@@ -96,29 +95,30 @@ async def run_ai_pipeline(log_id: str, tag_category_id: str, phrase: str):
|
|
|
try:
|
|
|
# step1: 正则过滤
|
|
|
result = await execute_reg(log_id,tag_category_id,phrase)
|
|
|
- logging.info(f"[{log_id}] Regex filtering result: {result}")
|
|
|
+ logger.info(f"[{log_id}] Regex filtering result: {result}")
|
|
|
|
|
|
# step2: 向量检索
|
|
|
if result:
|
|
|
result = vector_similarity_search(phrase, result)
|
|
|
- logging.info(f"[{log_id}] Vector search result: {result}")
|
|
|
+ logger.info(f"[{log_id}] Vector search result: {result}")
|
|
|
|
|
|
# step3: LLM 打标
|
|
|
if result:
|
|
|
result = await reflect_check(phrase, result)
|
|
|
result = [r.dict() for r in result.labels]
|
|
|
result = json.dumps(result, ensure_ascii=False)
|
|
|
- logging.info(f"[{log_id}] LLM result: {result}")
|
|
|
+ logger.info(f"[{log_id}] LLM result: {result}")
|
|
|
|
|
|
# step4: 更新数据库
|
|
|
- update_tag_log(log_id, str(result))
|
|
|
+ # 如果result是个空集合，插入None
|
|
|
+ update_tag_log(log_id, result if result else None)
|
|
|
|
|
|
except Exception as e:
|
|
|
- logging.error(f"[{log_id}] Pipeline failed: {e}")
|
|
|
+ logger.error(f"[{log_id}] Pipeline failed: {e}")
|
|
|
|
|
|
@router.post("/tagging")
|
|
|
async def ai_tagging(request: TaggingRequest,background_tasks: BackgroundTasks):
|
|
|
- logging.info(f"app_id: {request.app_id}, timestamp: {request.timestamp}, sign: {request.sign}, business_attr: {request.business_attr}, phrase: {request.phrase}")
|
|
|
+ logger.info(f"app_id: {request.app_id}, timestamp: {request.timestamp}, sign: {request.sign}, business_attr: {request.business_attr}, phrase: {request.phrase}")
|
|
|
# 数据库中插入一条记录，记录请求的app_id、timestamp、business_attr、phrase等信息，状态设为“处理中”，后续步骤完成后更新状态和结果
|
|
|
id = init_tag_log(request)
|
|
|
# 执行异步任务
|
|
|
@@ -128,7 +128,7 @@ async def ai_tagging(request: TaggingRequest,background_tasks: BackgroundTasks):
|
|
|
tag_category_id=request.tag_category_id,
|
|
|
phrase=request.phrase
|
|
|
)
|
|
|
- logging.info(f"Started background task for log_id: {id}")
|
|
|
+ logger.info(f"Started background task for log_id: {id}")
|
|
|
return {
|
|
|
"code": 200,
|
|
|
"message": "submit successful"
|
|
|
@@ -136,10 +136,11 @@ async def ai_tagging(request: TaggingRequest,background_tasks: BackgroundTasks):
|
|
|
|
|
|
@router.get("/query")
|
|
|
def query(business_attr: str):
|
|
|
+ logger.info(f"Querying tag log for business_attr: {business_attr}")
|
|
|
# 查询指定business_attr对应的标签信息
|
|
|
sql = """SELECT * FROM ai_tagging.aitag_tag_log WHERE business_attr = %s and is_delete = 0"""
|
|
|
result = dao.query_dict(sql, (business_attr,))
|
|
|
- print(result)
|
|
|
+ logger.info(f"Query result: {result}")
|
|
|
return {"code": 200, "message": "AI Query Endpoint", "data": result[0] if result else None}
|
|
|
|
|
|
class FeedbackRequest(BaseModel):
|
|
|
@@ -154,7 +155,7 @@ class FeedbackRequest(BaseModel):
|
|
|
|
|
|
@router.post("/feedback")
|
|
|
def ai_feedback(feedback_request: FeedbackRequest):
|
|
|
- logging.info(f"Received feedback request: {feedback_request}")
|
|
|
+ logger.info(f"Received feedback request: {feedback_request}")
|
|
|
# 这里将用户的反馈信息保存到数据库中aitag_tag_log，供后续分析和模型优化使用
|
|
|
dao.execute(
|
|
|
"""update ai_tagging.aitag_tag_log set feedback = %s, feedback_result = %s, feedback_time = %s, feedback_user_id = %s, feedback_user_nm = %s, contract_no = %s, feedback_user_org = %s, feedback_user_endpoint = %s, state = %s where business_attr = %s""",
|
jiayongqiang