|
|
@@ -16,6 +16,8 @@ import json
|
|
|
from agent.logger import logger
|
|
|
from agent.core.config import get_config_path
|
|
|
import asyncio
|
|
|
+from agent.core.tagging_state import TAGGING_STATE
|
|
|
+import time
|
|
|
|
|
|
config = get_config_path()
|
|
|
TOP_K = config['app']['top_k']
|
|
|
@@ -31,6 +33,7 @@ class TaggingRequest(BaseModel):
|
|
|
business_attr: str = Field(..., description="业务属性")
|
|
|
phrase: str = Field(..., description="需要打标签的文本")
|
|
|
tag_category_id: Optional[str] = Field(None, description="指定标签类别ID，默认为空表示不指定")
|
|
|
+ esb_seq_no: Optional[str] = Field(None,description="ESB流水号")
|
|
|
|
|
|
async def execute_reg(log_id:str,tag_category_id:str,phrase: str)-> list:
|
|
|
sql = f"""select
|
|
|
@@ -73,7 +76,7 @@ def vector_similarity_search(phrase: str, ids:list)-> list:
|
|
|
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]
|
|
|
r = [{"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]
|
|
|
- logger.info(f"{phrase} Vector search result: {r}")
|
|
|
+ # logger.info(f"{phrase} Vector search result: {r}")
|
|
|
return r
|
|
|
|
|
|
def init_tag_log(request: TaggingRequest):
|
|
|
@@ -86,20 +89,36 @@ def init_tag_log(request: TaggingRequest):
|
|
|
# 业务编号如果以test开头，则tag_scope = 1,否则都是0
|
|
|
tag_scope = 1 if request.business_attr.startswith("test") else 0
|
|
|
dao.execute(
|
|
|
- """INSERT INTO aitag_tag_log (id,app_id, insert_time, business_attr, phrase, state, tag_scope) VALUES (%s, %s, %s, %s, %s, %s, %s)""",
|
|
|
- (id,request.app_id, datetime.now(), request.business_attr, request.phrase, 0, tag_scope)
|
|
|
+ """INSERT INTO aitag_tag_log (id,app_id, insert_time, business_attr, phrase, state, tag_scope,esb_seq_no) VALUES (%s, %s, %s, %s, %s, %s, %s,%s)""",
|
|
|
+ (id,request.app_id, datetime.now(), request.business_attr, request.phrase, TAGGING_STATE.REQUEST.value, tag_scope, request.esb_seq_no)
|
|
|
)
|
|
|
return id
|
|
|
|
|
|
-def update_tag_log(id:str, result:str):
|
|
|
+def end_tagging(id:str, result:str):
|
|
|
dao.execute(
|
|
|
"""UPDATE aitag_tag_log SET state = %s, result = %s, ai_result_endtime = %s WHERE id = %s""",
|
|
|
- (1, result, datetime.now(), id)
|
|
|
+ (TAGGING_STATE.END.value, result, datetime.now(), id)
|
|
|
)
|
|
|
|
|
|
+def fail_tagging(id:str):
|
|
|
+ dao.execute(
|
|
|
+ """UPDATE aitag_tag_log SET state = %s, ai_result_endtime = %s WHERE id = %s""",
|
|
|
+ (TAGGING_STATE.FAIL.value, datetime.now(), id)
|
|
|
+ )
|
|
|
+
|
|
|
+def start_tagging(id:str):
|
|
|
+ dao.execute(
|
|
|
+ """UPDATE aitag_tag_log SET state = %s, ai_result_starttime = %s WHERE id = %s""",
|
|
|
+ (TAGGING_STATE.BEGIN.value, datetime.now(), id)
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
async def run_ai_pipeline(log_id: str, tag_category_id: str, phrase: str):
|
|
|
try:
|
|
|
async with background_semaphore:
|
|
|
+ logger.info(f"开始打标:{log_id}, {phrase}")
|
|
|
+ # step0: 开始打标
|
|
|
+ start_tagging(log_id)
|
|
|
# step1: 正则过滤
|
|
|
result = await execute_reg(log_id,tag_category_id,phrase)
|
|
|
# step2: 向量检索
|
|
|
@@ -111,17 +130,22 @@ async def run_ai_pipeline(log_id: str, tag_category_id: str, phrase: str):
|
|
|
except Exception as e:
|
|
|
logger.error(f"LLM reflection check failed: {e}")
|
|
|
result = None
|
|
|
+ fail_tagging(log_id)
|
|
|
+ return
|
|
|
# step4: 更新数据库
|
|
|
# 如果result是个空集合，插入None
|
|
|
- update_tag_log(log_id, result if result else None)
|
|
|
+ end_tagging(log_id, result if result else None)
|
|
|
|
|
|
except Exception as e:
|
|
|
logger.error(f"[{log_id}] Pipeline failed: {e}")
|
|
|
- update_tag_log(log_id, None)
|
|
|
+ fail_tagging(log_id)
|
|
|
+
|
|
|
|
|
|
+# 0:请求已接收；1：打标完成； 2：客户经理已经确认；3，结果已推送；
|
|
|
+# 4：开始打标， 5：打标失败
|
|
|
@router.post("/tagging")
|
|
|
async def ai_tagging(request: TaggingRequest,background_tasks: BackgroundTasks):
|
|
|
- logger.info(f"app_id: {request.app_id}, timestamp: {request.timestamp}, sign: {request.sign}, business_attr: {request.business_attr}, phrase: {request.phrase}")
|
|
|
+ logger.info(f"esb_seq_no: {request.esb_seq_no}, business_attr: {request.business_attr}, phrase: {request.phrase}")
|
|
|
# 数据库中插入一条记录，记录请求的app_id、timestamp、business_attr、phrase等信息，状态设为“处理中”，后续步骤完成后更新状态和结果
|
|
|
id = init_tag_log(request)
|
|
|
# 执行异步任务
|
|
|
@@ -162,7 +186,7 @@ def ai_feedback(feedback_request: FeedbackRequest):
|
|
|
# 这里将用户的反馈信息保存到数据库中aitag_tag_log，供后续分析和模型优化使用
|
|
|
dao.execute(
|
|
|
"""update 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""",
|
|
|
- (feedback_request.feedback, feedback_request.feedback_result, datetime.now(), feedback_request.user_id, feedback_request.user_nm, feedback_request.contract_no, feedback_request.user_org, feedback_request.user_endpoint, 2, feedback_request.business_attr)
|
|
|
+ (feedback_request.feedback, feedback_request.feedback_result, datetime.now(), feedback_request.user_id, feedback_request.user_nm, feedback_request.contract_no, feedback_request.user_org, feedback_request.user_endpoint, TAGGING_STATE.FEEDBACK.value, feedback_request.business_attr)
|
|
|
)
|
|
|
return {"code": 200, "message": "Feedback received successfully"}
|
|
|
|
jiayongqiang