添加ai日志

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="创建时间")

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>任意内容</think>，用于剥离推理模型的思考过程
 _THINK_RE = re.compile(r"<think>.*?</think>", 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)

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)

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