diff --git a/app/models/pg/ai_call_log.py b/app/models/pg/ai_call_log.py new file mode 100644 index 0000000..c22a519 --- /dev/null +++ b/app/models/pg/ai_call_log.py @@ -0,0 +1,22 @@ +"""PostgreSQL: ai_call_log 表模型""" + +from datetime import datetime +from typing import Optional + +from sqlalchemy import BigInteger, DateTime, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.database import PgBase + + +class AiCallLog(PgBase): + """AI 调用日志表""" + + __tablename__ = "ai_call_log" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, comment="主键ID(雪花)") + scene: Mapped[str] = mapped_column(String(32), nullable=False, comment="场景标识: structure/major_match/skill_extract/company_enrich") + system_prompt: Mapped[str] = mapped_column(Text, nullable=False, comment="系统提示词") + user_message: Mapped[str] = mapped_column(Text, nullable=False, comment="发给AI的用户消息") + response: Mapped[Optional[str]] = mapped_column(Text, comment="AI返回原文") + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="创建时间") diff --git a/app/services/ai_tool.py b/app/services/ai_tool.py index 658b4d0..14f07db 100644 --- a/app/services/ai_tool.py +++ b/app/services/ai_tool.py @@ -7,14 +7,19 @@ import asyncio import re -from typing import Any +from datetime import datetime +from typing import Any, Optional from json_repair import repair_json from langchain_core.language_models import BaseChatModel from langchain_core.messages import SystemMessage, HumanMessage +from snowflake import SnowflakeGenerator from app.core.logger import log +# AI 日志专用雪花ID +_log_id_gen = SnowflakeGenerator(instance=2) + # 匹配 任意内容,用于剥离推理模型的思考过程 _THINK_RE = re.compile(r".*?", re.DOTALL | re.IGNORECASE) @@ -39,7 +44,7 @@ def parse_llm_json(text: str) -> Any: return repair_json(cleaned, return_objects=True) -async def ai_chat(llm: BaseChatModel, system_prompt: str, user_message: str) -> str: +async def ai_chat(llm: BaseChatModel, system_prompt: str, user_message: str, scene: Optional[str] = None) -> str: """异步调用 LLM,返回原始文本。接口异常时触发门闸阻塞重试直到成功。""" global _gate_probing @@ -53,7 +58,11 @@ async def ai_chat(llm: BaseChatModel, system_prompt: str, user_message: str) -> HumanMessage(content=user_message), ] response = await llm.ainvoke(messages) - return response.content + content = response.content + # 记录 AI 调用日志(失败不影响主流程) + if scene: + await _save_call_log(scene, system_prompt, user_message, content) + return content except Exception as e: # 接口失败 → 关闭门闸 if _gate_event.is_set(): @@ -64,6 +73,9 @@ async def ai_chat(llm: BaseChatModel, system_prompt: str, user_message: str) -> # 当前协程成为探针,负责退避重试直到成功 _gate_probing = True result = await _probe_until_recover(llm, system_prompt, user_message) + # 记录 AI 调用日志 + if scene: + await _save_call_log(scene, system_prompt, user_message, result) # 探针成功,直接返回结果(不用再调一次) return result # 已有探针在工作,回到 while 顶部 await _gate_event.wait() 等待唤醒 @@ -96,9 +108,9 @@ async def _probe_until_recover(llm: BaseChatModel, system_prompt: str, user_mess delay = min(delay * 2, max_delay) -async def ai_chat_json(llm: BaseChatModel, system_prompt: str, user_message: str) -> Any: +async def ai_chat_json(llm: BaseChatModel, system_prompt: str, user_message: str, scene: Optional[str] = None) -> Any: """异步调用 LLM,返回解析后的 JSON 对象""" - raw = await ai_chat(llm, system_prompt, user_message) + raw = await ai_chat(llm, system_prompt, user_message, scene=scene) if not raw or not raw.strip(): log.warning("AI 返回为空") return None @@ -107,3 +119,26 @@ async def ai_chat_json(llm: BaseChatModel, system_prompt: str, user_message: str except Exception as e: log.warning("AI JSON 解析失败: {}, raw={}", e, raw[:200]) return None + + +async def _save_call_log(scene: str, system_prompt: str, user_message: str, response: Optional[str]) -> None: + """异步写入 AI 调用日志到 PG,失败只 warning 不影响主流程""" + try: + from app.core.database import PgSession + from sqlalchemy import insert + from app.models.pg.ai_call_log import AiCallLog + + async with PgSession() as pg: + await pg.execute( + insert(AiCallLog).values( + id=next(_log_id_gen), + scene=scene, + system_prompt=system_prompt, + user_message=user_message, + response=response, + created_at=datetime.now(), + ) + ) + await pg.commit() + except Exception as e: + log.warning("AI 调用日志写入失败: {}", e) diff --git a/app/services/company_clean_service.py b/app/services/company_clean_service.py index 2ff482e..0d2de32 100644 --- a/app/services/company_clean_service.py +++ b/app/services/company_clean_service.py @@ -62,7 +62,7 @@ async def _do_clean(company: dict) -> None: short_name = company.get("short_name", "") user_msg = f"【公司简称】\n{short_name}\n\n【行业列表】\n{dict_cache.industry_text}" - result = await ai_chat_json(CompanyCleanModel.ENRICH, COMPANY_ENRICH_SYSTEM, user_msg) + result = await ai_chat_json(CompanyCleanModel.ENRICH, COMPANY_ENRICH_SYSTEM, user_msg, scene="company_enrich") if result is None or not result.get("valid", False): await _update_status(company_id, 4) diff --git a/app/services/job_clean_service.py b/app/services/job_clean_service.py index 07d870e..888ed11 100644 --- a/app/services/job_clean_service.py +++ b/app/services/job_clean_service.py @@ -127,7 +127,7 @@ async def _do_clean(data: dict) -> None: # 第一次AI:结构化提取 user_message = _build_user_message(data) - result = await ai_chat_json(JobCleanModel.STRUCTURE, JOB_STRUCTURE_SYSTEM, user_message) + result = await ai_chat_json(JobCleanModel.STRUCTURE, JOB_STRUCTURE_SYSTEM, user_message, scene="structure") if result is None or not result.get("valid", False): log.info("[id={}] 丢弃:AI判定无效", data_id) await _update_pg_status(data_id, "discarded") @@ -217,7 +217,7 @@ async def _match_major(job_id: int, result: dict) -> None: req = result.get("requirement", "") user_msg = f"【岗位信息】\n标题: {title}\n职责: {desc}\n要求: {req}\n\n【专业分类列表】\n{dict_cache.major_category_text}" - data = await ai_chat_json(JobCleanModel.MAJOR_MATCH, MAJOR_MATCH_SYSTEM, user_msg) + data = await ai_chat_json(JobCleanModel.MAJOR_MATCH, MAJOR_MATCH_SYSTEM, user_msg, scene="major_match") if data is None: return @@ -242,7 +242,7 @@ async def _extract_skill_tags(job_id: int, result: dict) -> None: req = result.get("requirement", "") user_msg = f"【岗位信息】\n标题: {title}\n职责: {desc}\n要求: {req}" - skills = await ai_chat_json(JobCleanModel.SKILL_EXTRACT, SKILL_EXTRACT_SYSTEM, user_msg) + skills = await ai_chat_json(JobCleanModel.SKILL_EXTRACT, SKILL_EXTRACT_SYSTEM, user_msg, scene="skill_extract") if not skills or not isinstance(skills, list): return