diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..f06722a --- /dev/null +++ b/.dockerignore @@ -0,0 +1,8 @@ +venv/ +__pycache__/ +*.pyc +.env +data/ +.superpowers/ +.git/ +tests/ diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..5884bd3 --- /dev/null +++ b/.env.example @@ -0,0 +1,27 @@ +# AI模型配置 +AI_PROVIDER=glm +ZHIPU_API_KEY=your_zhipu_api_key_here +AI_MODEL=glm-4-flash +# 请求地址: 留空=用SDK默认(标准套餐); coding套餐用 https://open.bigmodel.cn/api/coding/paas/v4 +ZHIPU_BASE_URL=https://open.bigmodel.cn/api/coding/paas/v4 + +# 数据库 +DATABASE_URL=sqlite+aiosqlite:///./data/cc_ai_kefu.db + +# 邮件推送 +SMTP_HOST=smtp.example.com +SMTP_PORT=465 +SMTP_USER=your_email@example.com +SMTP_PASS=your_email_password +NOTIFY_EMAIL=notify@example.com + +# 询盘推送 UniService +PUSH_ENABLED=false +PUSH_API_URL=https://uniservice.server.hnxinyouhe.com/open/v1/inquiry/push +PUSH_API_KEY=your_uniservice_api_key_here +PUSH_EXTERNAL_KEY_PREFIX=ai-kefu + +# 应用配置 +APP_HOST=0.0.0.0 +APP_PORT=8016 +APP_DEBUG=true diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d348235 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +venv/ +__pycache__/ +*.pyc +*.pyo +.env +data/ +.superpowers/ +*.db diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..50b8153 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,17 @@ +FROM python:3.13-slim + +WORKDIR /app + +# Install dependencies first (layer caching) +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application code +COPY . . + +# Create data directory with correct permissions +RUN mkdir -p /app/data && chmod 777 /app/data + +EXPOSE 8000 + +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/__init__.py b/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/routes/__init__.py b/app/api/routes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/routes/admin.py b/app/api/routes/admin.py new file mode 100644 index 0000000..cf20457 --- /dev/null +++ b/app/api/routes/admin.py @@ -0,0 +1,218 @@ +import hashlib +import secrets +from datetime import datetime, timezone + +from fastapi import APIRouter, Cookie, Request, Response +from sqlalchemy import delete, select, func +from sqlalchemy.orm import load_only +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.main import async_session +from app.models.database import ( + Conversation, ConversationStatus, Message, CustomerProfile, Evaluation, IntentLevel, + LeadPushLog, +) +from app.services.ip_service import resolve_country + +router = APIRouter(prefix="/api/admin") + +_sessions: dict[str, float] = {} +SESSION_TTL = 86400 + + +def _check_auth(session_token: str | None) -> bool: + if not session_token or session_token not in _sessions: + return False + if _sessions[session_token] < datetime.now(timezone.utc).timestamp() - SESSION_TTL: + _sessions.pop(session_token, None) + return False + return True + + +@router.post("/login") +async def login(request: Request, response: Response): + body = await request.json() + password = body.get("password", "") + if password != settings.ADMIN_PASSWORD: + return {"ok": False, "error": "密码错误"} + token = secrets.token_hex(16) + _sessions[token] = datetime.now(timezone.utc).timestamp() + response.set_cookie("admin_token", token, httponly=True, max_age=SESSION_TTL) + return {"ok": True} + + +@router.post("/logout") +async def logout(response: Response, admin_token: str | None = Cookie(None)): + if admin_token and admin_token in _sessions: + _sessions.pop(admin_token) + response.delete_cookie("admin_token") + return {"ok": True} + + +@router.get("/conversations") +async def list_conversations( + admin_token: str | None = Cookie(None), + intent: str | None = None, + page: int = 1, + size: int = 20, +): + if not _check_auth(admin_token): + return {"error": "unauthorized"} + + async with async_session() as db: + # Only show conversations that have at least 1 message + msg_subq = ( + select(Message.conversation_id, func.count().label("mc"), + func.min(Message.content).filter(Message.role == "user").label("first_user_msg")) + .group_by(Message.conversation_id) + .subquery() + ) + + query = ( + select(Conversation, msg_subq.c.mc, msg_subq.c.first_user_msg) + .join(msg_subq, msg_subq.c.conversation_id == Conversation.id) + .order_by(Conversation.created_at.desc()) + ) + + if intent and intent in ("interested", "not_interested", "pending"): + query = query.join(Evaluation, Evaluation.conversation_id == Conversation.id, isouter=True) + query = query.where(Evaluation.intent_level == intent) + + total_q = select(func.count()).select_from(msg_subq) + total_result = await db.execute(total_q) + total = total_result.scalar() or 0 + + result = await db.execute(query.offset((page - 1) * size).limit(size)) + rows = result.all() + + items = [] + for conv, msg_count, first_msg in rows: + client_info = conv.client_info or {} + ip = client_info.get("ip", "") + country = resolve_country(ip) + + eval_result = await db.execute( + select(Evaluation).where(Evaluation.conversation_id == conv.id) + ) + evaluation = eval_result.scalar_one_or_none() + + preview = (first_msg or "")[:50] + + items.append({ + "id": conv.id, + "session_id": conv.session_id, + "status": conv.status.value, + "source": conv.source.value, + "ip": ip, + "device": client_info.get("user_agent", ""), + "country": country, + "intent_level": evaluation.intent_level.value if evaluation else None, + "reasoning": evaluation.reasoning if evaluation else None, + "contact_info": evaluation.contact_info if evaluation else None, + "email_sent": evaluation.email_sent if evaluation else False, + "message_count": msg_count, + "preview": preview, + "created_at": conv.created_at.isoformat() if conv.created_at else None, + }) + + return {"total": total, "page": page, "size": size, "items": items} + + +@router.get("/push-logs") +async def list_push_logs( + admin_token: str | None = Cookie(None), + status: str | None = None, + conversation_id: str | None = None, + page: int = 1, + size: int = 20, +): + """询盘推送日志(设计 2026-09-28-lead-push-to-uniservice §9)。 + + 列表不返回 request_payload(含完整对话,体积大)——排障看库。 + """ + if not _check_auth(admin_token): + return {"error": "unauthorized"} + + async with async_session() as db: + # load_only: 列表不含 request_payload(含完整对话,体积大)—— + # 直接不加载该列,省一页 ~20 条大 JSON 的 I/O 与反序列化 + query = select(LeadPushLog).options(load_only( + LeadPushLog.id, LeadPushLog.evaluation_id, + LeadPushLog.conversation_id, LeadPushLog.push_type, + LeadPushLog.status, LeadPushLog.attempts, LeadPushLog.outcome, + LeadPushLog.response_status, LeadPushLog.error_message, + LeadPushLog.created_at, LeadPushLog.updated_at, + )).order_by(LeadPushLog.created_at.desc()) + if status in ("pending", "success", "failed", "skipped"): + query = query.where(LeadPushLog.status == status) + if conversation_id: + query = query.where(LeadPushLog.conversation_id == conversation_id) + + total_result = await db.execute( + select(func.count()).select_from(query.subquery())) + total = total_result.scalar() or 0 + + result = await db.execute(query.offset((page - 1) * size).limit(size)) + logs = result.scalars().all() + + items = [{ + "id": log.id, + "evaluation_id": log.evaluation_id, + "conversation_id": log.conversation_id, + "push_type": log.push_type, + "status": log.status, + "attempts": log.attempts, + "outcome": log.outcome, + "response_status": log.response_status, + "error_message": log.error_message, + "created_at": log.created_at.isoformat() if log.created_at else None, + "updated_at": log.updated_at.isoformat() if log.updated_at else None, + } for log in logs] + + return {"total": total, "page": page, "size": size, "items": items} + + +@router.get("/conversations/{conv_id}/messages") +async def get_messages(conv_id: str, admin_token: str | None = Cookie(None)): + if not _check_auth(admin_token): + return {"error": "unauthorized"} + + async with async_session() as db: + result = await db.execute( + select(Message) + .where(Message.conversation_id == conv_id, Message.content != "") + .order_by(Message.created_at) + ) + messages = result.scalars().all() + + profile_result = await db.execute( + select(CustomerProfile).where(CustomerProfile.conversation_id == conv_id) + ) + profile = profile_result.scalar_one_or_none() + + return { + "messages": [ + { + "role": m.role.value, + "content": m.content, + "time": m.created_at.isoformat() if m.created_at else None, + } + for m in messages if m.content and m.content.strip() + ], + "profile": profile.raw_info if profile else {}, + } + + +@router.delete("/conversations/{conv_id}") +async def delete_conversation(conv_id: str, admin_token: str | None = Cookie(None)): + if not _check_auth(admin_token): + return {"error": "unauthorized"} + + async with async_session() as db: + await db.execute(delete(Evaluation).where(Evaluation.conversation_id == conv_id)) + await db.execute(delete(CustomerProfile).where(CustomerProfile.conversation_id == conv_id)) + await db.execute(delete(Message).where(Message.conversation_id == conv_id)) + await db.execute(delete(Conversation).where(Conversation.id == conv_id)) + await db.commit() + return {"ok": True} diff --git a/app/api/routes/chat.py b/app/api/routes/chat.py new file mode 100644 index 0000000..68e3f5b --- /dev/null +++ b/app/api/routes/chat.py @@ -0,0 +1,102 @@ +import json +import logging + +from fastapi import APIRouter, WebSocket, WebSocketDisconnect + +from app.main import async_session +from app.models.database import ConversationSource, ConversationStatus +from app.services.conversation import ConversationManager + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.websocket("/ws/chat") +async def ws_chat(websocket: WebSocket): + await websocket.accept() + session_id = None + + try: + async with async_session() as db: + manager = ConversationManager(db) + + # Collect client info + client_ip = websocket.client.host if websocket.client else "" + forwarded = websocket.headers.get("x-forwarded-for", "") + if forwarded: + client_ip = forwarded.split(",")[0].strip() + user_agent = websocket.headers.get("user-agent", "") + client_info = {"ip": client_ip, "user_agent": user_agent} + + # First message: init or resume + init_data = await websocket.receive_text() + init_msg = json.loads(init_data) + + if "session_id" in init_msg: + conv = await manager.get_by_session(init_msg["session_id"]) + if conv and conv.status == ConversationStatus.active: + session_id = conv.session_id + await websocket.send_text(json.dumps({ + "type": "resumed", + "session_id": session_id, + })) + else: + session_id = None + + # page_token: reuse active conversation from same page session + if not session_id and "page_token" in init_msg and init_msg["page_token"]: + conv = await manager.get_by_page_token(init_msg["page_token"]) + if conv: + session_id = conv.session_id + await websocket.send_text(json.dumps({ + "type": "resumed", + "session_id": session_id, + })) + + if not session_id: + source = ConversationSource( + init_msg.get("source", "standalone") + ) + merged_info = {**client_info, **(init_msg.get("client_info") or {})} + conv = await manager.create_conversation( + source=source, + client_info=merged_info, + page_token=init_msg.get("page_token"), + ) + session_id = conv.session_id + await websocket.send_text(json.dumps({ + "type": "init", + "session_id": session_id, + })) + + # Chat loop + while True: + data = await websocket.receive_text() + msg = json.loads(data) + content = msg.get("content", "").strip() + if not content: + continue + + ended = False + + async for event in manager.stream_process_message(session_id, content): + if event["type"] == "error": + await websocket.send_text(json.dumps({ + "type": "error", + "message": event["message"], + })) + break + + await websocket.send_text(json.dumps(event, ensure_ascii=False)) + + if event["type"] == "evaluated" and event.get("ended"): + ended = True + break + + if ended: + break + + except WebSocketDisconnect: + logger.info("Client disconnected: %s", session_id) + except Exception: + logger.exception("WebSocket error for session %s", session_id) diff --git a/app/api/routes/chat_rest.py b/app/api/routes/chat_rest.py new file mode 100644 index 0000000..221700d --- /dev/null +++ b/app/api/routes/chat_rest.py @@ -0,0 +1,142 @@ +"""REST shell for the HESU app (non-streaming, JSON-in/JSON-out). + +These routes wrap `ConversationManager`, shared with the existing `/ws/chat` +widget path. The app uses HTTP because it lacks a WebSocket client and is fine +with a 3-8s "AI is typing" affordance covering the round trip. + +Red lines (see spec): +- WS path (`app/api/routes/chat.py`) shares `ConversationManager` with these + routes; both paths stay drift-free. +- `context` seeding is conditional + whitelist (handled in + `ConversationManager.create_conversation`). +- The message route consumes `stream_process_message`; lead email is + triggered inside `ConversationManager` when contact info is detected + (contact-driven-lead-email plan). +""" +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel +from sqlalchemy import select + +from app.main import async_session +from app.models.database import ConversationSource, ConversationStatus, Message +from app.services.conversation import ConversationManager +from app.services.rate_limit import check as rate_check + +router = APIRouter(prefix="/api/chat", tags=["chat-rest"]) + + +class SessionRequest(BaseModel): + session_id: str | None = None + context: dict | None = None + client_info: dict | None = None + + +@router.post("/session") +async def chat_session(req: SessionRequest): + """Resume an active conversation by session_id, or create a new app one. + + `context` flows into `create_conversation` where it is whitelisted into + `raw_info` (only when non-None). `client_info` is stored for analytics only + (ip/ua/device-id/lang) and never fed to the AI prompt. + """ + async with async_session() as db: + manager = ConversationManager(db) + if req.session_id: + conv = await manager.get_by_session(req.session_id) + if conv and conv.status == ConversationStatus.active: + return { + "session_id": conv.session_id, + "is_new": False, + "ended": False, + } + conv = await manager.create_conversation( + source=ConversationSource.app, + client_info=req.client_info, + context=req.context, + ) + return { + "session_id": conv.session_id, + "is_new": True, + "ended": False, + } + + +@router.get("/{session_id}/messages") +async def chat_history(session_id: str): + """Public history endpoint for session replay. App-source only, to narrow + the read surface; an unknown or non-app session yields 404. + """ + async with async_session() as db: + manager = ConversationManager(db) + conv = await manager.get_by_session(session_id) + if not conv or conv.source != ConversationSource.app: + raise HTTPException(status_code=404, detail="session not found") + result = await db.execute( + select(Message) + .where(Message.conversation_id == conv.id) + .order_by(Message.created_at) + ) + msgs = result.scalars().all() + return [ + { + "role": m.role.value, + "content": m.content, + "created_at": m.created_at.isoformat(), + } + for m in msgs + ] + + +class MessageRequest(BaseModel): + session_id: str + content: str + + +@router.post("/message") +async def chat_message(req: MessageRequest, request: Request): + """Process one app turn: stream-collect the AI reply, then evaluate. + + 🔴 Consumes `stream_process_message` so the REST path shares one code + path with the WS widget and never drifts. Lead email is triggered inside + `ConversationManager` when contact info is detected + (contact-driven-lead-email plan) — this route carries no email logic. + + A light in-memory rate limit (per ip + X-Device-Id, 30/h) guards GLM + spend; only this route is throttled. + """ + ip = request.client.host if request.client else "" + dev = request.headers.get("X-Device-Id", "") + if not rate_check(ip, dev): + raise HTTPException(status_code=429, detail="rate limit exceeded") + + async with async_session() as db: + manager = ConversationManager(db) + conv = await manager.get_by_session(req.session_id) + if not conv: + raise HTTPException(status_code=404, detail="session not found") + if conv.status != ConversationStatus.active: + raise HTTPException(status_code=409, detail="conversation ended") + + reply_parts: list[str] = [] + ended = False + intent_level = "pending" + async for event in manager.stream_process_message( + req.session_id, req.content + ): + t = event.get("type") + if t == "stream_chunk": + reply_parts.append(event.get("content", "")) + elif t == "evaluated": + ended = bool(event.get("ended")) + intent_level = event.get("intent_level", "pending") + elif t == "error": + msg = event.get("message", "error") + raise HTTPException( + status_code=409 if "ended" in msg else 404, detail=msg + ) + + return { + "reply": "".join(reply_parts), + "ended": ended, + "intent_level": intent_level, + } diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..041db6d --- /dev/null +++ b/app/config.py @@ -0,0 +1,54 @@ +from pydantic_settings import BaseSettings +from pathlib import Path + + +class Settings(BaseSettings): + APP_HOST: str = "0.0.0.0" + APP_PORT: int = 8000 + APP_DEBUG: bool = False + + AI_PROVIDER: str = "glm" + ZHIPU_API_KEY: str = "" + # 留空=用 SDK 默认地址(标准套餐 https://open.bigmodel.cn/api/paas/v4) + # coding 套餐填 https://open.bigmodel.cn/api/coding/paas/v4 + ZHIPU_BASE_URL: str = "" + AI_MODEL: str = "glm-4-flash" + + DATABASE_URL: str = "sqlite+aiosqlite:///./data/cc_ai_kefu.db" + + REDIS_URL: str = "redis://localhost:6379/0" + REDIS_PASSWORD: str = "" + SESSION_TTL: int = 86400 # 24 hours + + SMTP_HOST: str = "" + SMTP_PORT: int = 465 + SMTP_USER: str = "" + SMTP_PASS: str = "" + NOTIFY_EMAIL: str = "" + + # 询盘推送 UniService(docs/superpowers/specs/2026-09-28-lead-push-to-uniservice-design.md) + PUSH_ENABLED: bool = False + PUSH_API_URL: str = "https://uniservice.server.hnxinyouhe.com/open/v1/inquiry/push" + PUSH_API_KEY: str = "" + PUSH_EXTERNAL_KEY_PREFIX: str = "ai-kefu" + PUSH_TIMEOUT_SECONDS: float = 10.0 + PUSH_MAX_ATTEMPTS: int = 3 + PUSH_RETRY_BACKOFF_BASE: float = 1.0 # 尝试间隔 1s/2s(3 次尝试、2 次间隔) + + ADMIN_PASSWORD: str = "admin123" + + model_config = { + "env_file": ".env", + "env_file_encoding": "utf-8", + } + + +BASE_DIR = Path(__file__).resolve().parent.parent +DATA_DIR = BASE_DIR / "data" +STATIC_DIR = BASE_DIR / "static" + +settings = Settings() + + +def ensure_dirs(): + DATA_DIR.mkdir(exist_ok=True) diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..1f76809 --- /dev/null +++ b/app/main.py @@ -0,0 +1,75 @@ +from contextlib import asynccontextmanager + +from fastapi import FastAPI +from fastapi.staticfiles import StaticFiles +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.config import settings, ensure_dirs, STATIC_DIR +from app.models.database import Base + + +# 后台推送任务的独立连接写结果可能撞上 SMTP 窗口内持有的写锁,默认 5s +# 忙等上限低于 SMTP 最坏时长,抬到 30s;等待发生在 aiosqlite 工作线程,不阻塞事件循环 +engine = create_async_engine(settings.DATABASE_URL, echo=settings.APP_DEBUG, + connect_args={"timeout": 30}) +async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + +async def _migrate(conn): + result = await conn.execute(text("PRAGMA table_info(conversations)")) + columns = {row[1] for row in result} + if "page_token" not in columns: + await conn.execute(text( + "ALTER TABLE conversations ADD COLUMN page_token VARCHAR(64)" + )) + + result = await conn.execute(text("PRAGMA table_info(evaluations)")) + columns = {row[1] for row in result} + if "correction_count" not in columns: + await conn.execute(text( + "ALTER TABLE evaluations ADD COLUMN correction_count INTEGER DEFAULT 0" + )) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + ensure_dirs() + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + await _migrate(conn) + # 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed + # 函数内导入: lead_push_service 顶部 from app.main import async_session, + # 顶层互相导入会循环 + from app.services.lead_push_service import sweep_stale_pending + await sweep_stale_pending() + yield + + +app = FastAPI(title="AI汽车客服", lifespan=lifespan) + +from app.api.routes.chat import router as chat_router +from app.api.routes.admin import router as admin_router +from app.api.routes.chat_rest import router as chat_rest_router +app.include_router(chat_router) +app.include_router(admin_router) +app.include_router(chat_rest_router) + +app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") + + +@app.get("/") +async def index(): + from fastapi.responses import FileResponse + return FileResponse(str(STATIC_DIR / "chat.html")) + + +@app.get("/admin") +async def admin_page(): + from fastapi.responses import FileResponse + return FileResponse(str(STATIC_DIR / "admin.html")) + + +@app.get("/health") +async def health(): + return {"status": "ok"} diff --git a/app/models/__init__.py b/app/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/models/database.py b/app/models/database.py new file mode 100644 index 0000000..1ef4aa2 --- /dev/null +++ b/app/models/database.py @@ -0,0 +1,161 @@ +import enum +import uuid +from datetime import datetime, timezone + +from sqlalchemy import String, Text, Boolean, Integer, ForeignKey, JSON, Enum as SAEnum +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship + + +class Base(DeclarativeBase): + pass + + +class ConversationStatus(str, enum.Enum): + active = "active" + completed = "completed" + abandoned = "abandoned" + + +class MessageRole(str, enum.Enum): + user = "user" + assistant = "assistant" + system = "system" + + +class ConversationSource(str, enum.Enum): + standalone = "standalone" + embedded = "embedded" + widget = "widget" + app = "app" + + +class IntentLevel(str, enum.Enum): + interested = "interested" + not_interested = "not_interested" + pending = "pending" + + +class Conversation(Base): + __tablename__ = "conversations" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + session_id: Mapped[str] = mapped_column(String(64), unique=True, index=True) + page_token: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + status: Mapped[ConversationStatus] = mapped_column( + SAEnum(ConversationStatus), default=ConversationStatus.active + ) + source: Mapped[ConversationSource] = mapped_column( + SAEnum(ConversationSource), default=ConversationSource.standalone + ) + client_info: Mapped[dict | None] = mapped_column(JSON, nullable=True) + created_at: Mapped[datetime] = mapped_column( + default=lambda: datetime.now(timezone.utc) + ) + updated_at: Mapped[datetime] = mapped_column( + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + ) + + messages: Mapped[list["Message"]] = relationship( + back_populates="conversation", order_by="Message.created_at" + ) + profile: Mapped["CustomerProfile | None"] = relationship( + back_populates="conversation", uselist=False + ) + evaluation: Mapped["Evaluation | None"] = relationship( + back_populates="conversation", uselist=False + ) + + +class Message(Base): + __tablename__ = "messages" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + conversation_id: Mapped[str] = mapped_column( + ForeignKey("conversations.id", ondelete="CASCADE") + ) + role: Mapped[MessageRole] = mapped_column(SAEnum(MessageRole)) + content: Mapped[str] = mapped_column(Text) + created_at: Mapped[datetime] = mapped_column( + default=lambda: datetime.now(timezone.utc) + ) + + conversation: Mapped["Conversation"] = relationship(back_populates="messages") + + +class CustomerProfile(Base): + __tablename__ = "customer_profiles" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + conversation_id: Mapped[str] = mapped_column( + ForeignKey("conversations.id", ondelete="CASCADE"), unique=True + ) + raw_info: Mapped[dict | None] = mapped_column(JSON, nullable=True, default=dict) + summary: Mapped[str | None] = mapped_column(Text, nullable=True) + updated_at: Mapped[datetime] = mapped_column( + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + ) + + conversation: Mapped["Conversation"] = relationship(back_populates="profile") + + +class Evaluation(Base): + __tablename__ = "evaluations" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + conversation_id: Mapped[str] = mapped_column( + ForeignKey("conversations.id", ondelete="CASCADE"), unique=True + ) + intent_level: Mapped[IntentLevel] = mapped_column(SAEnum(IntentLevel)) + reasoning: Mapped[str | None] = mapped_column(Text, nullable=True) + contact_info: Mapped[dict | None] = mapped_column(JSON, nullable=True) + email_sent: Mapped[bool] = mapped_column(Boolean, default=False) + correction_count: Mapped[int] = mapped_column(Integer, default=0) + evaluated_at: Mapped[datetime] = mapped_column( + default=lambda: datetime.now(timezone.utc) + ) + + conversation: Mapped["Conversation"] = relationship(back_populates="evaluation") + + +class LeadPushLog(Base): + """询盘推送 UniService 日志(设计 2026-09-28-lead-push-to-uniservice)。 + + status: pending → success/failed;skipped = contact 三选一(phone/whatsapp/ + email)不满足,对方必拒,预检后不发 HTTP 只留审计痕迹。 + outcome: 200 时对方逐条结论 accepted/duplicated(duplicated 不是错误)。 + """ + __tablename__ = "lead_push_logs" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + evaluation_id: Mapped[str] = mapped_column( + ForeignKey("evaluations.id", ondelete="CASCADE"), index=True + ) + conversation_id: Mapped[str] = mapped_column( + ForeignKey("conversations.id", ondelete="CASCADE"), index=True + ) + push_type: Mapped[str] = mapped_column(String(16)) # lead/returning/correction + status: Mapped[str] = mapped_column(String(16), default="pending", index=True) + attempts: Mapped[int] = mapped_column(Integer, default=0) + request_payload: Mapped[dict | None] = mapped_column(JSON, nullable=True) + response_status: Mapped[int | None] = mapped_column(Integer, nullable=True) + outcome: Mapped[str | None] = mapped_column(String(16), nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column( + default=lambda: datetime.now(timezone.utc) + ) + updated_at: Mapped[datetime] = mapped_column( + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + ) diff --git a/app/models/schemas.py b/app/models/schemas.py new file mode 100644 index 0000000..e835f1e --- /dev/null +++ b/app/models/schemas.py @@ -0,0 +1,57 @@ +from datetime import datetime + +from pydantic import BaseModel + +from app.models.database import ConversationStatus, ConversationSource, MessageRole, IntentLevel + + +class MessageCreate(BaseModel): + role: MessageRole + content: str + + +class MessageOut(BaseModel): + id: str + role: MessageRole + content: str + created_at: datetime + + model_config = {"from_attributes": True} + + +class ConversationCreate(BaseModel): + source: ConversationSource = ConversationSource.standalone + client_info: dict | None = None + + +class ConversationOut(BaseModel): + id: str + session_id: str + status: ConversationStatus + source: ConversationSource + client_info: dict | None + created_at: datetime + messages: list[MessageOut] = [] + + model_config = {"from_attributes": True} + + +class CustomerProfileOut(BaseModel): + raw_info: dict | None + summary: str | None + + model_config = {"from_attributes": True} + + +class EvaluationOut(BaseModel): + intent_level: IntentLevel + reasoning: str | None + contact_info: dict | None + email_sent: bool + + model_config = {"from_attributes": True} + + +class ChatMessage(BaseModel): + session_id: str + content: str diff --git a/app/prompts/company_facts.md b/app/prompts/company_facts.md new file mode 100644 index 0000000..2cb13fb --- /dev/null +++ b/app/prompts/company_facts.md @@ -0,0 +1,29 @@ +# 公司事实卡(仅供回答客户关于公司的事实性问题) + +## 我是谁 +- Hesu Auto 是 Hesu Group 旗下专注汽车出口的公司 +- 中国首批汽车出口企业之一,累计出口 50,000+ 台 +- **累计出口到 70+ 国家与地区**(中东、非洲、南美、欧洲)——出口目的国口径,不是分支/网点覆盖 +- 与 30+ 汽车品牌深度合作:BYD、Changan、Geely、Leopard、ROX、Nissan、Zeeker、Li Auto 等 + +## 卖什么 +- **新车与二手车都做** +- 全动力类型:燃油、混动、插混、纯电 +- 全车型类别:轿车、SUV、皮卡、重卡;乘用车、商用车、新能源车、车队/项目采购 + +## 网络与运作 +- 模式:中国采购供应 → 迪拜(Jebel Ali)枢纽备货分拨 → 海外交付 +- 海外分支五处:Dubai、Ghana、Ethiopia、Djibouti、Belarus +- Abu Dhabi 有展车节点;Ghana 西非保税仓;Djibouti+Ethiopia 东非协同;Venezuela(南美)在网络节点内 + +## 使用规则(与 system.md 红线同级) +- **统一兜底(最高优先)**:任何不清楚、事实卡没有、需要核实的问题——**不编造、不承诺、不否定、不断言**,统一动作只有一个:顺势引导留联系方式转客户经理——"这个我让我们经理给你准确答复——留个 WhatsApp,他直接联系你" +- **不扩展原则**:回答公司事实**只用本卡原词**——禁止添加城市/地址/街道等任何卡外细节(卡片没写城市就不提城市)。卡片没写的就是不知道,一律走统一兜底 +- 客户问公司情况/品牌/是否可信时,取上述事实**简短作答(1-2 句)**,建立信任 +- **分支问题与出口覆盖问题是两回事,分开答**: + - 问分支/网点("do you have branch in X")→ 只认卡里列的:五分支+节点(Dubai/Ghana/Ethiopia/Djibouti/Belarus/Abu Dhabi/Venezuela)必须答"有",**只说国家+节点类型,不提城市**,可逐字使用:"Yes, we've got a Ghana branch with bonded storage";卡里没有的 → **走统一兜底**——绝对禁止编造否定答案("Not yet"/"we don't have"/"没有") + - 问出口覆盖("do you ship to X")→ 70+ 国是**累计出口目的国**,不代表 X 一定有网点:答"已出口到 70+ 国家"事实 + **走统一兜底**(X 具体情况经理确认)——不承诺、不否定 +- 绝不主动长篇介绍公司,绝不用成营销腔——风格仍按【说话风格】 +- 涉及具体价格/交付时效/库存承诺 → 仍按【职能边界】转客户经理,本卡不改变任何红线;售后只说"有本地协同支持",不做任何保证 +- 事实卡没写的细节(某国代理、某车型现车、认证资质细节)→ **走统一兜底**(不编造,留资+经理沟通) +- 数字一律照抄本卡写法(如 50,000+、70+),不改写成中文数字 diff --git a/app/prompts/evaluator.md b/app/prompts/evaluator.md new file mode 100644 index 0000000..74a6d84 --- /dev/null +++ b/app/prompts/evaluator.md @@ -0,0 +1,44 @@ +你是一个客户意向评估助手。根据对话中收集到的客户信息,判断这个客户的购买意向。 + +## 评估标准 + +### interested(有意向) +必须满足以下所有条件: +- 明确表达了购车需求(有预算或目标车型/品牌) +- 明确了目标国家/地区或采购用途 +- 提供了至少一种联系方式(手机/WhatsApp/邮箱/微信) +- 对话至少进行了3轮以上(客户说了3条以上消息) +- 客户表现出真实采购意向而非简单咨询 +- 客户询问价格、优惠、运费或试驾/交付安排,应视为意向积极信号(有真实购买兴趣) +- 客户主动索要我方联系方式(WhatsApp/邮箱/电话),视为最强意向信号之一 + +### not_interested(无意向) +客户表现出以下特征: +- 明确表示不买、只是随便看看 +- 态度冷淡,不愿多聊 +- 表示已经从其他渠道购买了 + +### pending(待定) +以下情况都应标记为pending: +- 对话轮次少于3轮 +- 还未收集到联系方式 +- 客户信息不够充分(缺少预算、车型、数量等关键信息) +- 客户还在比较或犹豫阶段 + +## 严格规则 +1. info_sufficient 必须为 true 时,intent_level 才可以是 interested 或 not_interested +2. 如果对话轮次少于3轮,info_sufficient 必须为 false,intent_level 必须为 pending +3. 如果没有联系方式,info_sufficient 必须为 false + +## 输出格式 +返回JSON,包含以下字段: +- intent_level: "interested" 或 "not_interested" 或 "pending" +- reasoning: 你的判断理由(1-2句话) +- contact_info: 联系方式对象,包含手机、WhatsApp、邮箱、微信,没有的留空字符串 +- info_sufficient: true或false,表示当前信息是否足够做出评估 + +## 客户信息 +$profile + +## 最近对话 +$recent_messages diff --git a/app/prompts/extractor.md b/app/prompts/extractor.md new file mode 100644 index 0000000..b25d0cd --- /dev/null +++ b/app/prompts/extractor.md @@ -0,0 +1,19 @@ +你是一个信息提取助手。请从以下对话历史中提取客户相关信息,输出JSON格式。 + +## 输出格式 +返回一个JSON对象,包含你能从对话中识别到的所有信息字段。字段名用中文。示例字段: +目标国家/地区、采购用途、预算、意向品牌、意向车型、采购数量、车况要求、年份要求、运输方式、联系方式、其他备注。 + +其中联系方式如果有的话格式为:{"手机": "xxx", "WhatsApp": "xxx", "邮箱": "xxx", "微信": "xxx"} +- 如果对话中客户更正了此前提供的联系方式(如 wrong number / 号码错了 / use this / 换成 / 又给了一个新号),联系方式必须以客户最后提供的值为准覆盖旧值 +- 如果客户在消息中直接发送了一串数字(整条消息只有号码),必须用该号码更新联系方式 + +## 规则 +- 只提取对话中明确提到的信息,不要猜测 +- 如果某个信息没有提到,不要包含该字段 +- 已有的信息请保留,新信息追加或更新 +- 之前的提取结果:$existing_info +- 如果没有之前的提取结果,$existing_info 为空对象 + +## 对话历史 +$messages diff --git a/app/prompts/system.md b/app/prompts/system.md new file mode 100644 index 0000000..bd5280f --- /dev/null +++ b/app/prompts/system.md @@ -0,0 +1,179 @@ +你是一位汽车出口外贸销售(新车与二手车都做),日常工作就是通过聊天和海外客户沟通。你说话像在 WhatsApp 上跟朋友聊天一样,简短、直接、有人情味。 + +## 语言规则(最高优先级) +- 客户用什么语言,你就用什么语言回复,严格匹配,绝不混用 +- 默认英语 + +## 职能边界(最高优先级,仅语言规则除外) + +你的准则只有一条:**需求了解 + 留资**——介绍车、挖意向、收联系方式。 +你不承担销售职能:不做任何承诺(价格/时间/库存)、不说绝对化的话、不代表公司做决定。 +涉及价格、时效和实际行动的事,一律由客户经理承接——你的任务是把需求和联系方式收好,交出去。 + +### 三类请求的处理 +1. **汽车信息**(车型/配置/参数/续航/年份/库存/汽车知识):直接热情介绍。 + - 公司情况(品牌/覆盖国家/出口经验/运作模式)按系统提供的事实卡简短作答建立信任——但不因此报价、承诺或跳过留资引导 +2. **价格、优惠、折扣、运费、关税、交付时效类**:一律不直接回答,用迂回话术: + - "Prices depend on the config, destination and shipping — our manager will prepare an accurate quote for you. What's your WhatsApp or email?" + - "Delivery time depends on your destination and shipping route — leave your contact and our manager will give you an exact schedule." + - 问价时若已知意向车型:同一回复内"确认车型 + 引导留资"一气呵成——带车型信息的线索才有价值 + - 零上下文问价(快捷按钮首触):先问需求(车型/国家),不急着要号 +3. **实质请求**(试驾、验车、下单、锁车、预留库存、售后、"帮我安排"类,以及经销/批发合作谈价格与条款):一律转接: + - "I'll connect you with our manager to arrange that — drop your WhatsApp and they'll reach out directly." + +### 客户报预算时 +- 理解预算档位,据此推荐合适的车型 +- 但绝对不要复述预算数字,也不要报任何车价 +- 确认预算只用无数字表述:"That budget works well" / "Got your budget"—— + 绝不说 "20K / 20000 / a budget of 20K" 这类任何数字形式(含改写形式如 20k) +- 反问意向加深了解:"SUV or sedan? How many seats do you need?" +- 然后顺势引导留资:"Tell me your WhatsApp and our manager will send you an accurate quote for your budget." + +### 节奏豁免(解决与"3轮规则"的冲突) +- 客户**主动**问价、问时效、提实质请求或索要我方联系方式时,价格红线立即生效(当轮就不给数字),不受"至少聊3轮"限制 +- 但**留资时机仍看需求了解程度**: + - 已知基本需求(客户提过车型/国家)→ 当轮即可迂回引导留资 + - 零上下文点击询价按钮(如 Get Quotation 首条消息)→ **先用一轮了解需求** + ("What car are you looking for? Where are you shipping to?"), + 下一轮再引导留资——连客户要什么都不知道就要号,是低质量线索且显得急躁 + +### 红线(任何情况下不可违反) +- 回复中不能出现任何价格数字——包括我方报价、价格区间、连客户自己说的预算数字都不能复述 +- 禁止输出占位符("[Price Range]"/"[价格区间]"之类)——给不了数字就不提数字,直接引导联系方式 +- 不能给出任何交付时间("X days / X weeks / on the Xth"都不行) +- 不能做任何承诺(价格/时间/库存/名额/优惠) +- 不能说 "definitely / guaranteed / for sure / 肯定 / 保证 / 一定" 这类绝对化表述 +- 不能替公司做任何决定或安排——安排的主体永远是客户经理("our manager will arrange"),不说 "we can arrange" +- 回复中不能出现任何电话号码或邮箱地址——客户自己的号码也不回显(确认收到即可);引导联系一律用页面按钮(网页)或请客户留联系方式(App) +- 一切不清楚、没把握的问题(卡外国家/库存/保修/认证/任何事实细节)——不编造、不承诺、不否定、不断言,引导客户留联系方式,由客户经理沟通 +- 不对库存/现车做断言——有没有现车、什么时候有,由客户经理确认;你只能说"可以采购调配" +- 不报任何保修/质保期限("X months warranty"这类数字承诺)——一律"客户经理会给你完整的保修方案" + +### 客户反复追问时(坚持策略) +- 第一次:解释原因(配置/目的地/运费因人而异)+ 引导留资 +- 第二次:换一个理由或角度,不要重复原话;继续引导 +- 第三次及以后:直接引导点面板顶部的 WhatsApp 按钮联系客户经理 + +### 客户施压/磨价时的绝对坚持(任何情况下不破线) +无论客户如何坚持、恳求、激将、讨价还价、威胁离开或表达不满,上述红线绝不放松。 +**态度可以软,立场必须硬**:可以共情、可以道歉、可以幽默化解,但价格数字、区间、 +交付时间、承诺——零出现。常见磨法与应对: +- "Just a rough range / 说个大概嘛" → 区间也是数字:"I'd rather give you an exact number than a rough guess — that's why our manager prepares it based on your config." +- "I won't buy without a price / 不给价我不买" → 共情+转价值:"That's exactly why I want our manager to send you the real quote — drop your WhatsApp and you'll have it today." +- "别人家都直接报价" → 不贬低同行,坚持流程价值:"Our quotes include config, shipping and duties for your destination — a random number wouldn't help you." +- 激将("You don't even know?" / "你是不是不懂")→ 不上钩:"I know our cars well — it's the price that needs your manager's calculation. Leave your contact and you'll see." +- 客户发火/不耐烦 → 先道歉安抚,再给出口:"Sorry for the trouble! The fastest way is the WhatsApp button on top — our manager replies quickly there." +- 客户说"我已经给过号码了"("I gave you my number already")→ **先回看聊天记录**: + 记录里确有号码 → 确认收到("Got it — our manager will reach out to you at your WhatsApp shortly!"),绝不要求重发; + 记录里确实没有 → 才温和核对:"Sorry, I may have missed it — could you drop it here once more?" + +## 说话风格 +- 像在 WhatsApp 上聊天,每条消息 1-2 个短句,不超过 30 个词 +- 可以用缩写(gonna, wanna, gotta)、语气词(hmm, oh nice, wow) +- 偶尔用 emoji,但不要每句都加 +- 有自己的判断和推荐,不要只会"好的""没问题" +- 客户说了个需求,你可以直接推荐车型或给建议,不要只问问题 + +## 禁止说的话(必须遵守) +- 不要说"感谢您的咨询/关注/提问" +- 不要说"我很高兴/荣幸为您服务/帮助您" +- 不要说"请随时联系我"或"不要犹豫" +- 不要说"希望以上信息对您有帮助" +- 不要说"如果您有任何其他问题" +- 不要用"首先...其次...最后..."这种列举式回答 +- 不要重复客户说的话("您提到您想买一辆丰田"这种废话) +- 不要总结对话内容 +- 不要每次回复都带问号结尾 + +## 对话推进规则(最关键,违反就是致命bug) +- 客户已经告诉你的信息,绝对不能再问一遍。客户说了车型就不要再问"什么车",说了年份就不要再问"哪年" +- 客户给出具体需求后,必须给出有用的回应(车型推荐、可否采购调配、配置对比),而不是继续追问 +- 对话必须向前推进,每条回复都要在"需求了解 + 留资"上更进一步 +- 如果客户连续两次被问到已经回答过的问题,客户会直接走人 +- 涉及价格/交付时效/实质安排的请求,按【职能边界】处理——这不是拒绝客户,而是把他交给能给出准确答案的人 + +## 上下文追踪(最关键,违反就是致命bug) +- 每条回复前,先回顾整个对话历史,明确客户当前想要什么 +- 客户的短消息(如"2024"、"how much"、"yes")必须结合之前的上下文理解,不能当作独立问题 +- 如果客户之前说了具体车型,后续所有回复都必须围绕这个车型,直到客户自己换了话题 +- 绝对不能在客户指定了车型之后推荐完全不同的车型(客户说BYD,你推荐Toyota,客户会立刻关闭页面) + +## 好回复 vs 差回复 +差:"感谢您的咨询!我们公司专注于二手车出口业务多年,拥有丰富的经验。请问您的目标市场是哪个国家呢?" +好:"Oh nice, what country are you looking to ship to?" + +差:"您好!很高兴为您服务。请问您的预算范围是多少呢?我们可以根据您的预算为您推荐合适的车型。" +好:"What's your budget range? I got some good options." + +差:"您提到的丰田卡罗拉是一款非常受欢迎的车型,我们在非洲和东南亚市场有大量现车。请问您对年份有什么要求吗?" +好:"Corolla is a solid pick, we can source it easily. Any preference on year?" + +差:客户说了"BYD Han 2025",AI回复"What kind of car are you looking for?" +好:客户说了"BYD Han 2025",AI回复"Nice, the 2025 Han comes in EV and DM-i versions. Which one you after?" + +差:客户说了年份"2025",AI回复"Any specific make or model in mind?" +好:客户说了年份"2025",AI回复"For 2025 models we can source BYD Han, Toyota Camry, Honda Accord. Budget range?" + +差:客户说了"BYD Tang",然后说"2024",AI回复"Got a 2024 model in mind? I can show you some great options!" +好:客户说了"BYD Tang",然后说"2024",AI回复"2024 Tang comes as EV 7-seater and DM-i AWD. Which one fits you?" + +差:客户说了"BYD Tang 2024",问"what is available",AI回复"Got Toyota Corolla, Honda Accord, BMW 3 Series" +好:客户说了"BYD Tang 2024",问"what is available",AI回复"The 2024 Tang comes as the EV 7-seater and DM-i AWD — we can source either. Want specs?" + +## 对话流程 +1. 开场轻松,简单打个招呼,问客户想找什么车 +2. 聊2-3轮了解基本需求(目标国家、预算、车型偏好) +3. 第3-4轮用"转接客户经理"或"发报价单"的方式引导联系方式(见下方策略) +4. 拿到联系方式后:确认一句("Got it!"),然后继续正常解答客户问题——不必急着结束对话,也别再提联系方式 + +## 联系方式引导策略(关键) +聊了3-4轮、对客户需求有基本了解后,用"转接客户经理"的方式要联系方式(客户**主动**问价/提实质请求时例外——见【职能边界】节奏豁免): + +### 英文话术示例 +- "Let me connect you with our manager who handles [country] market — what's your WhatsApp?" +- "I can have our team send you some quotes. Drop your WhatsApp or email?" +- "Want me to get you a proper quote? I'll need your WhatsApp so our manager can reach you." + +### 中文话术示例 +- "我让负责[国家]市场的同事加你,你WhatsApp多少?" +- "我让客户经理给你报个价,留个WhatsApp或者邮箱吧?" +- "你给我个联系方式,我让客户经理给你推几个车型。" + +### 引流兜底(先聊后引,顺序不能反) +(话术模板见【职能边界】) +- 聊天内按上述方式要联系方式,最多两次 +- 两次都没要到,才指引用户点面板上的 WhatsApp 按钮直接找我们: + "You can also just message us on WhatsApp (button at the top) — our guy replies fast." +- 禁止一上来就推 WhatsApp 按钮跳过聊天索要 + +### 客户索要我方联系方式(最高意向信号,立即响应) +客户问"how can I contact you / 你们的WhatsApp多少"时: +- 系统提醒会告知当前会话来源。**网页会话**(有页面按钮)→ 指引点击: + "Just tap the WhatsApp button at the top of this chat — our manager replies fast!" + (邮箱同理:"Tap the Email button at the top.") +- **App 会话**(没有页面按钮)→ 反向留资: + "Drop your WhatsApp and our manager will reach out to you directly with the full quote!" +- 中文版同理("直接点窗口顶部的 WhatsApp 按钮就行,客户经理回复很快~") +- 【互留】网页会话给完指路后可轻问一句:"Drop yours too and our manager will send you + the full quote!"——每会话最多一次;客户已留过或疑似留过则不加 +- 客户索要未提供的渠道(微信/Telegram/LINE/ins)→ "We mainly use WhatsApp — tap the + button at the top, fastest way to reach us.",绝不提供或编造任何渠道账号 + +### 系统提醒配合 +- 系统会在上下文提醒里告诉你客户是否已留联系方式 +- 提醒说"已提供"时:绝对禁止再要,确认收到即可 +- 提醒说"疑似提供"时:礼貌请客户确认一次,不要当成没给过重新索要 + +### 注意 +- 只要一次没要到,可以换个理由再试一次,但最多两次 +- 不要在开场就**主动**要联系方式,至少聊3轮(客户主动问价/提实质请求时例外,见职能边界) +- 拿到后说一句"Got it, our team will reach out to you soon"就行,别再重复要 + +## 关注的信息 +自然地在聊天中了解这些,不要逐条问: +- 客户所在国家/地区 +- 预算范围 +- 意向品牌/车型 +- 采购数量 +- 车况/年份要求 +- 联系方式(WhatsApp、邮箱等) diff --git a/app/providers/__init__.py b/app/providers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/providers/base.py b/app/providers/base.py new file mode 100644 index 0000000..0963688 --- /dev/null +++ b/app/providers/base.py @@ -0,0 +1,24 @@ +from abc import ABC, abstractmethod +from collections.abc import AsyncGenerator + + +class AIProvider(ABC): + @abstractmethod + async def chat(self, messages: list[dict], system_prompt: str) -> str: + """Send messages and return assistant reply text.""" + + @abstractmethod + async def stream_chat( + self, messages: list[dict], system_prompt: str + ) -> AsyncGenerator[str, None]: + """Stream assistant reply token by token.""" + + @abstractmethod + async def chat_with_extraction( + self, messages: list[dict], system_prompt: str + ) -> tuple[str, dict]: + """Send messages and return (reply_text, extracted_json). + + The provider should request JSON output and parse it. + Falls back to empty dict if parsing fails. + """ diff --git a/app/providers/glm.py b/app/providers/glm.py new file mode 100644 index 0000000..2dd1c0a --- /dev/null +++ b/app/providers/glm.py @@ -0,0 +1,94 @@ +import asyncio +import json +import logging +from collections.abc import AsyncGenerator + +from zhipuai import ZhipuAI + +from app.providers.base import AIProvider + +logger = logging.getLogger(__name__) + +DEFAULT_MODEL = "glm-4-flash" + + +class GLMProvider(AIProvider): + def __init__(self, api_key: str, model: str = "glm-4-flash", base_url: str = ""): + self.client = ZhipuAI(api_key=api_key, base_url=base_url or None) + self.model = model + + async def chat(self, messages: list[dict], system_prompt: str) -> str: + full_messages = [{"role": "system", "content": system_prompt}] + messages + return await self._call(full_messages, temperature=0.3) + + async def stream_chat( + self, messages: list[dict], system_prompt: str + ) -> AsyncGenerator[str, None]: + full_messages = [{"role": "system", "content": system_prompt}] + messages + + def _sync_stream(): + resp = self.client.chat.completions.create( + model=self.model, + messages=full_messages, + stream=True, + temperature=0.3, + ) + for chunk in resp: + delta = chunk.choices[0].delta + if delta and delta.content: + yield delta.content + + try: + loop = asyncio.get_event_loop() + gen = _sync_stream() + + def _next_token(g): + try: + return next(g) + except StopIteration: + return None + + while True: + token = await loop.run_in_executor(None, _next_token, gen) + if token is None: + break + yield token + except Exception: + logger.exception("GLM streaming call failed") + raise + + async def chat_with_extraction( + self, messages: list[dict], system_prompt: str + ) -> tuple[str, dict]: + full_messages = [{"role": "system", "content": system_prompt}] + messages + if not messages: + full_messages.append({"role": "user", "content": "请根据以上要求输出结果。"}) + + raw = await self._call(full_messages, temperature=0.1) + extracted = self._parse_json(raw) + return raw, extracted + + async def _call(self, messages: list[dict], temperature: float = 0.3) -> str: + def _sync_call(): + resp = self.client.chat.completions.create( + model=self.model, + messages=messages, + temperature=temperature, + ) + return resp.choices[0].message.content + + try: + return await asyncio.to_thread(_sync_call) + except Exception: + logger.exception("GLM API call failed") + raise + + @staticmethod + def _parse_json(text: str) -> dict: + try: + start = text.index("{") + end = text.rindex("}") + 1 + return json.loads(text[start:end]) + except (ValueError, json.JSONDecodeError): + logger.warning("Failed to parse JSON from model output") + return {} diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/ai_service.py b/app/services/ai_service.py new file mode 100644 index 0000000..bbccd33 --- /dev/null +++ b/app/services/ai_service.py @@ -0,0 +1,98 @@ +import json +import logging +from collections.abc import AsyncGenerator +from pathlib import Path + +from app.providers.base import AIProvider +from app.providers.glm import GLMProvider +from app.config import settings + +logger = logging.getLogger(__name__) + +PROMPTS_DIR = Path(__file__).resolve().parent.parent / "prompts" + + +def _load_prompt(filename: str) -> str: + return (PROMPTS_DIR / filename).read_text(encoding="utf-8") + + +def create_provider() -> AIProvider: + if settings.AI_PROVIDER == "glm": + return GLMProvider( + api_key=settings.ZHIPU_API_KEY, + model=settings.AI_MODEL, + base_url=settings.ZHIPU_BASE_URL, + ) + raise ValueError(f"Unknown AI provider: {settings.AI_PROVIDER}") + + +SYSTEM_PROMPT = _load_prompt("system.md") +COMPANY_FACTS = _load_prompt("company_facts.md") # 公司事实卡(知识与行为分离) +EXTRACTOR_PROMPT = _load_prompt("extractor.md") +EVALUATOR_PROMPT = _load_prompt("evaluator.md") + +# 对话基座: 行为 prompt 在前(红线显著性),事实卡居中(参考资料), +# 当轮 context 提醒最后拼接(在 chat/stream_chat 内,保持最新鲜) +_BASE_CHAT_PROMPT = SYSTEM_PROMPT + "\n\n" + COMPANY_FACTS + + +class AIService: + def __init__(self, provider: AIProvider | None = None): + self.provider = provider or create_provider() + + async def chat(self, messages: list[dict], context: str = "") -> str: + """Main conversation: get AI reply to user message.""" + prompt = _BASE_CHAT_PROMPT + ("\n\n" + context if context else "") + return await self.provider.chat(messages, prompt) + + async def stream_chat(self, messages: list[dict], context: str = "") -> AsyncGenerator[str, None]: + """Stream AI reply token by token.""" + prompt = _BASE_CHAT_PROMPT + ("\n\n" + context if context else "") + async for token in self.provider.stream_chat(messages, prompt): + yield token + + async def extract_info( + self, messages: list[dict], existing_info: dict + ) -> dict: + """Extract structured info from conversation history.""" + prompt = EXTRACTOR_PROMPT.replace( + "$existing_info", json.dumps(existing_info or {}, ensure_ascii=False) + ).replace("$messages", self._format_messages(messages)) + _, extracted = await self.provider.chat_with_extraction( + messages=[], system_prompt=prompt + ) + return extracted + + async def evaluate( + self, profile: dict, recent_messages: list[dict] + ) -> dict: + """Evaluate customer intent level.""" + prompt = EVALUATOR_PROMPT.replace( + "$profile", json.dumps(profile, ensure_ascii=False) + ).replace("$recent_messages", self._format_messages(recent_messages[-6:])) + _, result = await self.provider.chat_with_extraction( + messages=[], system_prompt=prompt + ) + return self._normalize_evaluation(result) + + @staticmethod + def _format_messages(messages: list[dict]) -> str: + lines = [] + for m in messages: + role = m.get("role", "unknown") + content = m.get("content", "") + label = {"user": "客户", "assistant": "顾问"}.get(role, role) + lines.append(f"{label}: {content}") + return "\n".join(lines) + + @staticmethod + def _normalize_evaluation(result: dict) -> dict: + intent = result.get("intent_level", "pending") + if intent not in ("interested", "not_interested", "pending"): + intent = "pending" + return { + "intent_level": intent, + "reasoning": result.get("reasoning", ""), + "contact_info": result.get("contact_info", {}), + "info_sufficient": result.get("info_sufficient", False), + } diff --git a/app/services/boundary_filter.py b/app/services/boundary_filter.py new file mode 100644 index 0000000..254d662 --- /dev/null +++ b/app/services/boundary_filter.py @@ -0,0 +1,199 @@ +"""职能边界输出过滤(方案B): AI 回复价格形态的确定性拦截。 +流式回复无法撤回,故在流出前全量缓冲过滤;违规整体替换为标准迂回话术。 +模式以 scripts/breach_rate.py 为基础并收严(预算复述/金融收严;公里等误报除外项)。 + +评审修正(13.6 复查): +- 货币组尾部 \\b 对中文备选(万/块钱/美元)永远无法命中——中文无词边界, + "2万左右"的 万与左 之间没有 \\b(实测漏检预算复述 6/6 的原形态)。 + \\b 只保留在拉丁备选 usd 之后(防 "usdroid" 类前缀误命中)。 +- 公里/km 排除必须直接挂在 万块? 之后: 挂在末尾裸 万 备选上会被 + 万块? 的空块匹配绕过("3万公里"仍命中,实测)。 +- 预算复述(裸 budget+数字/中文数字万)、金融裸数(首付/月供)收严; + 里程 k 形式(30k km/45k miles)除外。质保期不再豁免(政策变更 2026-09-10: + 保修期限是编造的公司政策承诺,一并墙级拦截,见 PRICE_PATTERNS 时间承诺组)。""" +import re + +from app.services.contact_pipeline import digits_only + +PRICE_PATTERNS = [ + re.compile(r"\$\s?\d"), + re.compile(r"\d[\d,.]*\s*(usd\b|美元|块钱|万块?(?!\s*(公里|km)))", re.I), + re.compile(r"\d[\d,.]*\s?k\b(?!\s*(km|kilometers?|kilometres?|miles?|mi\b|公里))", re.I), + re.compile(r"(price|cost|quote)[^.\d]{0,20}(? bool: + return any(p.search(reply or "") for p in PRICE_PATTERNS) + + +# ---- 13.7: 句子粒度门控 ---- +# 完整句切分: 英文句尾标点后需跟空白才切(小数 "2.5" 的句点后是数字,天然不切); +# 中文标点无需后续空白即可切。finditer 实现为无损——句间空白保留为下一片段的 +# 前导空白(方案稿的 re.split 会吃掉空白,实测把 "A. B" 粘成 "A.B", +# 干净回复逐句流出后无法逐字还原)。 +_SENT_BOUNDARY = re.compile(r"(?<=[.!?;;])\s+|(?<=[。!?…])") + + +def split_sentences(text: str) -> tuple[list[str], str]: + """切出完整句子,返回 (完整句子列表, 不完整尾巴)。 + 无损保证: "".join(完整) + 尾巴 == 原文。完整句不带尾随空白。""" + if not text: + return [], "" + parts: list[str] = [] + prev = 0 + for m in _SENT_BOUNDARY.finditer(text): + parts.append(text[prev:m.start()]) + prev = m.start() # 边界(句间空白)留在下一片段开头 + parts.append(text[prev:]) + parts = [p for p in parts if p] + if not parts: + return [], text + return parts[:-1], parts[-1] + + +# 过渡话术(pivot): 句子门控拦截时已流出干净前缀,接一句过渡收尾。 +# 与开场版迂回话术不同——读起来像回复中段的转向,故单条不轮换。 +PIVOTS_EN = [ + "Actually, for the exact price — what's your WhatsApp? Our manager will send it over.", +] +PIVOTS_ZH = [ + "具体的报价,客户经理会发给你——留个WhatsApp或邮箱?", +] +PIVOTS_EN_CONTACT = [ + "Our manager will send the exact price to your WhatsApp shortly.", +] +PIVOTS_ZH_CONTACT = [ + "客户经理稍后会把准确报价发到你的 WhatsApp。", +] + + +def pivot_for(user_msg: str, has_contact: bool = False) -> str: + zh = bool(re.search(r"[一-鿿]", user_msg or "")) + if has_contact: + pool = PIVOTS_ZH_CONTACT if zh else PIVOTS_EN_CONTACT + else: + pool = PIVOTS_ZH if zh else PIVOTS_EN + return pool[0] + + +def deflection_for(user_msg: str, rotate: int = 0, + has_contact: bool = False) -> str: + zh = bool(re.search(r"[一-鿿]", user_msg or "")) + if has_contact: + pool = DEFLECTIONS_ZH_CONTACT if zh else DEFLECTIONS_EN_CONTACT + else: + pool = DEFLECTIONS_ZH if zh else DEFLECTIONS_EN + return pool[rotate % len(pool)] + + +# ---- 15.3: 零号码墙(doctrine v2) ---- +# AI 绝不输出任何号码/邮箱: 与价格墙并列的第二道确定性红线。 +LONG_NUMBER_RE = re.compile(r"\+?\d[\d\s\-().]{9,17}\d") # ≥10位连续数字(允许分隔形态) +ANY_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}") + +NO_CONTACT_WEB = ("Just tap the WhatsApp button at the top of this chat — " + "our manager replies fast!") +NO_CONTACT_APP = ("Drop your WhatsApp and our manager will reach out to you directly " + "with the full quote!") +# F6 同款(15.3 补): 已留资时零号码墙替换句不得索要 +NO_CONTACT_ALREADY = ("Got it — our manager will reach out to you at your " + "WhatsApp shortly with the full quote!") + + +def no_contact_value(sentence: str, user_history: list[dict] | None = None) -> bool: + """零号码墙(v2): AI 输出不得含任何 ≥10 位数字串或邮箱。 + 唯一豁免: 该串出现在近期客户消息中(客户自己号码的礼貌回显)。 + 无值比对——年份/里程/VIN/续航天然安全(已验证)。""" + for m in LONG_NUMBER_RE.finditer(sentence or ""): + c = digits_only(m.group(0)) + if user_history and any(c in digits_only(msg.get("content") or "") + for msg in user_history if msg.get("role") == "user"): + continue # 客户回显豁免 + return True + for m in ANY_EMAIL_RE.finditer(sentence or ""): + if user_history and any(m.group(0).lower() in (msg.get("content") or "").lower() + for msg in user_history if msg.get("role") == "user"): + continue + return True + return False + + +# ---- Task 5b: 再索要墙(已留资时输出侧拦截) ---- +# 仅当调用方 has_contact=True 才启用(接线在 conversation.py); +# 每条模式强制渠道词在场,防泛请求误杀("Can you send me the details?" 放行)。 +# 模式集与 system.md 全部留资话术模板逐句对齐并实证(13 命中/12 放行)—— +# GLM 背的就是这些模板,模板全集必须全收。 +REASK_PATTERNS = [ + # what's/which's your <渠道词> + re.compile(r"(?:what|which)['′]?s\s+(?:your|the)\s+" + r"(?:whatsapp|number|phone|e-?mail|contact|号码|联系方式)", re.I), + # got/have your <渠道词> ... ?(问号兜尾——确认句 "Got your number, thanks!" 无问号放行) + re.compile(r"(?:got|have)\s+(?:your|the)\s+" + r"(?:whatsapp|number|phone|e-?mail|contact)[^!?.]{0,60}\?", re.I), + # 陈述式需求: (I'll) need/want your <渠道词>——"I'll need your WhatsApp" 无问号也要拦 + re.compile(r"(?:i['′]ll\s+|i\s+)?(?:need|want)\s+(?:your|the)\s+" + r"(?:whatsapp|number|phone|e-?mail|contact|号码|联系方式)", re.I), + # Tell me your <渠道词> + re.compile(r"tell\s+me\s+(?:your\s+)?" + r"(?:whatsapp|number|phone|e-?mail|contact|号码|联系方式)", re.I), + # could/can you + 给予动词 + ... 渠道词(动词表无 confirm/send——礼貌确认与泛请求放行) + re.compile(r"(?:could|can|would|may)\s+you\s+" + r"(?:leave|drop|share|give|provide|type|post)[^.!?\n]{0,30}" + r"(?:whatsapp|number|phone|e-?mail|contact|号码|联系方式|微信|邮箱)", re.I), + # 祈使: leave/drop/share/give (me) your/us a <渠道词> + re.compile(r"(?:leave|drop|share|give)\s+(?:me\s+)?(?:your|us\s+your?|a)\s+" + r"(?:whatsapp|contact|number|e-?mail|联系方式)", re.I), + # 互留后缀 "Drop yours too..."(无渠道词,单列) + re.compile(r"(?:drop|leave)\s+yours\b", re.I), + # "WhatsApp or email?" 独立短问 + re.compile(r"(?:whatsapp|e-?mail|contact)\s+or\s+(?:whatsapp|e-?mail|contact)\s*\?", re.I), + # 裸碎片问句: "Your number, please?"/"Your WhatsApp?"(Ghana 案例穿透补) + # 后顾排除跟进确认句("reach/got/sent/did your X?" 是问进度不是要号) + re.compile(r"(? bool: + """输出句是否在索要联系方式(仅在客户已留资时用于丢弃)。""" + return any(p.search(sentence or "") for p in REASK_PATTERNS) diff --git a/app/services/contact_pipeline.py b/app/services/contact_pipeline.py new file mode 100644 index 0000000..a47d755 --- /dev/null +++ b/app/services/contact_pipeline.py @@ -0,0 +1,264 @@ +"""统一联系方式管道:联系方式检测的唯一事实源。 + +三个动作: +- 归一化:中英文 key 别名统一为 phone/whatsapp/email/wechat +- 深合并:同 key 新值覆盖旧值(客户改号场景),空值不覆盖 +- 校验分级:confirmed(格式通过,可触发邮件/结束)/ suspect(疑似, + 仅用于提醒 AI 礼貌确认,防止重复索要——D6/S1) + +上游约定:raw_info 中 "_" 前缀 key 为内部状态,禁止进入提示词渲染。 +""" +import re + +KEY_ALIASES = { + "phone": "phone", "手机": "phone", "电话": "phone", "电话号码": "phone", + "联系电话": "phone", "手机号": "phone", "手机号码": "phone", + "tel": "phone", "telephone": "phone", "phone_number": "phone", + "phone number": "phone", "mobile": "phone", "mobile_number": "phone", + "mobile number": "phone", + "whatsapp": "whatsapp", "wa": "whatsapp", "wsp": "whatsapp", + "whats app": "whatsapp", "whatsapp账号": "whatsapp", + "whatsapp号码": "whatsapp", "whatsapp number": "whatsapp", + "email": "email", "邮箱": "email", "e-mail": "email", "mail": "email", + "电子邮件": "email", + "wechat": "wechat", "微信": "wechat", "wx": "wechat", "weixin": "wechat", +} + + +def normalize_contacts(raw: dict | None) -> dict: + """key 归一化 + 去空值。精确别名优先;未命中再做 contains 兜底 + (extractor 即兴变体 key 不静默丢值)。contains 按 KEY_ALIASES 声明序 + 取第一个命中——复合 key(电话/WhatsApp)结果确定。""" + out: dict[str, str] = {} + for k, v in (raw or {}).items(): + key = str(k).strip().lower() + alias = KEY_ALIASES.get(key) + if alias is None: + for name, target in KEY_ALIASES.items(): + if name in key: + alias = target + break + val = str(v or "").strip() + if alias and val: + out[alias] = val + return out + + +def merge_contacts(old: dict | None, new: dict | None) -> dict: + """同 key 新值覆盖旧值;空/缺省不覆盖(堵浅合并抹数据 bug)。""" + return {**normalize_contacts(old), **normalize_contacts(new)} + + +def digits_only(s: str) -> str: + """提取纯数字(split_validate 位数判断与后续 canon 复用)。""" + return re.sub(r"\D", "", s) + + +def split_validate(contacts: dict) -> tuple[dict, dict]: + """返回 (confirmed, suspect)。电话/WhatsApp 取数字 7-15 位为合格;微信取非空 ≥4 字符。""" + confirmed: dict[str, str] = {} + suspect: dict[str, str] = {} + for k, v in normalize_contacts(contacts).items(): + if k == "email": + ok = re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", v) is not None + elif k == "wechat": + ok = len(v.strip()) >= 4 # 微信号无数字格式,非空短串即可 + else: + ok = 7 <= len(digits_only(v)) <= 15 + (confirmed if ok else suspect)[k] = v + return confirmed, suspect + + +EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}") +INTL_PHONE_RE = re.compile(r"\+\d{6,15}") +# 邻接绑定:上下文词与数字必须在同一匹配内(仅空白衔接)——禁止"消息任意位置 +# 有上下文词 + 任意位置有 7 位数字"的消息级误绑(S1:车架号/订单号/先令价格) +WHATSAPP_CTX_RE = re.compile( + r"(?:whats\s*app|wa\.?me|(?:^|\b)wa\s*[::])\D{0,12}(\+?\d[\d\s\-().]{5,17}\d)", + re.IGNORECASE, +) +PHONE_CTX_RE = re.compile( + r"(?:my\s+(?:(?:new|other|another)\s+)?(?:phone\s+)?number\s+is|" + r"my\s+(?:(?:new|other|another)\s+)?phone\s+is|" + r"(?:msg|message|text|ping)\s+me\s+(?:at|on)?\s*|" + r"call\s+me\s+(?:at|on)|reach\s+me\s+(?:at|on)|" + r"phone\s*[::]|number\s*[::]|号码\s*[::]|手机号?\s*[::]|电话\s*[::])\s*" + r"(\+?\d[\d\s\-().]{5,17}\d)", + re.IGNORECASE, +) +# 仅 scan_user_messages 使用的裸邻接形态: "my number 0712345678"(无 is/冒号)。 +# ⚠️ 不得进 latest_stated_contacts——裸邻接无自述确定性,授覆盖权会让首轮旧值 +# 长期压制 extractor 后续改号(实证: test_correction_up_to_three 更正邮件全灭)。 +# 已知覆盖边界: "my phone 0712345678"(phone 变体,无 is)不在本形态内。 +PHONE_ADJACENCY_SCAN_RE = re.compile( + r"my\s+(?:(?:new|other|another)\s+)?number\s+(\+?\d[\d\s\-().]{5,17}\d)", + re.IGNORECASE, +) +# 倒装自述: "0712 345 678 is my whatsapp / 13800138000 我的微信" +# ⚠️ 只收显式渠道词(whatsapp/wa/wechat/微信),不收 number/phone/号码—— +# 自查实证: "2500000 is my number one choice" 这类排位表述会把价格误绑成 +# 号码,且本正则进入 latest_stated_contacts 即获覆盖权,误捕获会翻新掉 +# 已存真实号码(正是 S1 系测试在防的事故类)。倒装电话表述罕见,不值得 +# 为其承担该风险;电话主通路由 PHONE_CTX_RE 前缀形态/standalone/INTL 覆盖。 +# (?:my|我的)后用 \s* 而非 \s+: 中文"我的微信"连写无空白,\s+ 永不命中(实证)。 +NUMBER_FIRST_RE = re.compile( + r"(\+?\d[\d\s\-().]{5,17}\d)\s*(?:is|这是)?\s*" + r"(?:my|我的)\s*(whats\s*app|wa\b|wechat|微信)", + re.IGNORECASE, +) + +_NUMBER_FIRST_CHANNEL = { + "whatsapp": "whatsapp", "whats app": "whatsapp", "wa": "whatsapp", + "wechat": "wechat", "微信": "wechat", +} + + +def _number_first_channel(hit: re.Match) -> str: + # whats\s*app 可捕到任意空白变体(双空格/NBSP/换行),查表前先归一成 + # 单空格——否则 KeyError 会沿着 scan 的每消息热路径直接断流整轮对话。 + return _NUMBER_FIRST_CHANNEL[" ".join(hit.group(2).lower().split())] + + +# 更正语境(与裸国际号同消息出现时才授予覆盖权) +CORRECTION_CTX_RE = re.compile( + r"wrong\s+(?:number|phone)|typo|打错|写错|输错|错了|换成了?|改为|" + r"use\s+(?:this|new|the\s+new)|instead|correct\s+(?:number|phone|it)", + re.IGNORECASE, +) + + +def scan_user_messages(messages: list[dict]) -> dict: + """S4: 扫描全部客户消息,返回高置信捕获(email / +国际号 / whatsapp·wa.me / + 显式电话上下文邻接数字)。裸数字(价格/年份/里程/车架号)一律不收。""" + found: dict[str, str] = {} + for msg in messages: + if msg.get("role") != "user": + continue + text = msg.get("content", "") + if "email" not in found: + hit = EMAIL_RE.search(text) + if hit: + found["email"] = hit.group(0) + if "phone" not in found: + sn = standalone_number(text) + if sn: + found["phone"] = sn + if "phone" not in found: + hit = INTL_PHONE_RE.search(text) + if hit: + found["phone"] = hit.group(0) + if "whatsapp" not in found: + hit = WHATSAPP_CTX_RE.search(text) + if hit: + found["whatsapp"] = hit.group(1) + if "phone" not in found: + hit = PHONE_CTX_RE.search(text) + if hit: + found["phone"] = hit.group(1) + if "phone" not in found: + hit = PHONE_ADJACENCY_SCAN_RE.search(text) + if hit: + found["phone"] = hit.group(1) + # 倒装属最低置信层: whatsapp/phone 任一已被更高置信源捕获即跳过, + # 不与之竞争(跨 key 门控故含 phone)。 + if "whatsapp" not in found and "phone" not in found: + hit = NUMBER_FIRST_RE.search(text) + if hit: + ch = _number_first_channel(hit) + found[ch] = hit.group(1) + return found + + +def canon(value: str) -> str: + """去重比对用的规范形:邮箱小写,其余取纯数字(S3/R3 best-effort)。""" + value = str(value or "").strip() + if "@" in value: + return value.lower() + return digits_only(value) + + +def standalone_number(text: str) -> str | None: + """整条消息仅为一个号码(Task 12d): 客户"直接甩号"的最显式表达,授予覆盖权。 + ≥9 位下限把年份(4)/美元价(5)/先令价(5-7)全部排除在外; + 允许 +/-/空格/横线/点/括号包裹,其余任何字符(字母/逗号)即非独立号码条。""" + t = (text or "").strip() + if not re.fullmatch(r"\+?[\d\s\-().·]{9,25}", t): + return None + return t if 9 <= len(digits_only(t)) <= 15 else None + + +def latest_stated_contacts(messages: list[dict]) -> dict: + """最新显式自述(唯一可覆盖 extractor 回传的来源,Task 12): + + 同 key 取全历史最后一次显式自述。覆盖权仅授予: + - 上下文门控捕获(whatsapp/wa.me/wa:、my (new/other) number is、msg/text/ping me、colon 形式、email) + - 倒装渠道自述(NUMBER_FIRST_RE: "0712 345 678 is my whatsapp"、 + "13800138000 我的微信"——只认显式渠道词) + - "更正语境 ∧ 同消息裸国际号"(wrong number, use +86...) + 纯裸数字无覆盖权(S1): 防价格/年份/订单号误翻新。 + 例外(Task 12d): 整条消息仅为一个号码(≥9位) = 最显式自述,授予覆盖权。 + 通则(权威契约): 正则接进本循环即授覆盖权——新增低精度形态时只进 + 扫描辅源 scan_user_messages,不进这里(先例: PHONE_ADJACENCY_SCAN_RE, + 裸 my number 邻接曾致更正邮件全灭)。 + """ + latest: dict[str, str] = {} + for msg in messages: + if msg.get("role") != "user": + continue + text = msg.get("content") or "" + hit = EMAIL_RE.search(text) + if hit: + latest["email"] = hit.group(0) + hit = WHATSAPP_CTX_RE.search(text) + if hit: + latest["whatsapp"] = hit.group(1) + sn = standalone_number(text) + if sn: + latest["phone"] = sn + hit = PHONE_CTX_RE.search(text) + if hit: + latest["phone"] = hit.group(1) + elif CORRECTION_CTX_RE.search(text): + hit = INTL_PHONE_RE.search(text) + if hit: + latest["phone"] = hit.group(0) + hit = NUMBER_FIRST_RE.search(text) + if hit: + ch = _number_first_channel(hit) + latest[ch] = hit.group(1) + return latest + + +def resolve_by_recency(before: dict, after: dict, messages: list[dict]) -> dict: + """最近提及裁决(Task 12c): before 中同 canon 的非 email key 构成一个镜像身份组。 + 组内分歧时,取"出现在最晚客户消息里"的 canon 代表值—— + - 杀掉 extractor 在历史号码间的振荡(旧号总在更早的消息里,天然落败) + - 压制幻觉值(从未出现在任何消息里,last_idx=-1) + - 客户最新消息里的新号(extractor 抓到的裸号)天然胜出 + 组内一致或不同组(本就不同的号)时原样返回。""" + out = dict(after) + groups: dict[str, set[str]] = {} + for k, v in (before or {}).items(): + if k != "email" and v: + groups.setdefault(canon(str(v)), set()).add(k) + user_digits = [ + digits_only(m.get("content") or "") for m in messages + if m.get("role") == "user" + ] + + def last_idx(c: str) -> int: + for i in range(len(user_digits) - 1, -1, -1): + if c and c in user_digits[i]: + return i + return -1 + + for keys in groups.values(): + vals = {k: out[k] for k in keys if out.get(k)} + distinct = {canon(str(v)) for v in vals.values()} + if len(distinct) > 1: + best = max(distinct, key=last_idx) + if last_idx(best) >= 0: + rep = next(v for v in vals.values() if canon(str(v)) == best) + for k in keys: + out[k] = rep + return out diff --git a/app/services/conversation.py b/app/services/conversation.py new file mode 100644 index 0000000..25bfc5f --- /dev/null +++ b/app/services/conversation.py @@ -0,0 +1,645 @@ +import logging +import re +import uuid +from collections.abc import AsyncGenerator +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.database import ( + Conversation, ConversationStatus, ConversationSource, + Message, MessageRole, CustomerProfile, Evaluation, IntentLevel, + LeadPushLog, +) +from app.services import email_service, lead_push_service +from app.services.ai_service import AIService +from app.services.boundary_filter import ( + NO_CONTACT_ALREADY, NO_CONTACT_APP, NO_CONTACT_WEB, asks_contact, + deflection_for, no_contact_value, pivot_for, split_sentences, + violates_boundary, +) +from app.services.contact_pipeline import ( + canon, latest_stated_contacts, merge_contacts, normalize_contacts, + resolve_by_recency, scan_user_messages, split_validate, +) + +logger = logging.getLogger(__name__) + +# 职能边界触发词(Task 13.2, 评审#1/#4 修订): 命中即注入红线提醒。精修说明: +# - 裸 "how long"/"多久" 不收(续航类问题常用),英文要求 take/arrive/deliver/ship 邻接, +# 中文只收 到货/多久能到/交货 等明确交付语境 +# - 【评审#1】施压话术本身也触发: 磨价轮("just roughly"/"说个大概")没有价格词, +# 却是最易破防的时刻——压力词单列一组 +# - 【评审#4】v1 补齐漏词: expensive/fee/贵/价钱/总价/落地/划算/免费(短语式)等; +# cost/fee 用 \b 边界,costa rica 不再误命中("costa"含cost但无词边界) +# - 原则: 误触发低害(多一条提醒),漏触发高危(红线失守),词表偏宽;上线后漏词案例回流补词 +BOUNDARY_TRIGGER_RE = re.compile( + r"price|pricing|how\s+much|\bcost\b|discount|cheaper|expensive|\bfee\b|" + r"quote|quotation|shipping\s+(fee|cost|time)|freight|tariff|customs\s+fee|" + r"best\s+offer|any\s+offer|free\s+shipping|" + r"test\s+drive|book(ing)?\s|reserv|arrange|" + r"delivery|lead\s+time|arriv|" + r"how\s+long\s+(does|will|would|to).*(take|arriv|deliver|ship)|" + r"roughly|ballpark|just\s+tell|come\s+on|at\s+least\s+(a|some|give)|" + r"budget|预算|价格|价钱|多少钱|报价|优惠|折扣|便宜|贵|划算|" + r"总价|落地|运费|关税|免费|" + r"到货|多久能到|交货|时效|发货|试驾|预约|下单|定金|提车|安排|" + r"大概|估个|估一|随便说|说个|求你|求了", + re.IGNORECASE, +) + +# 预算路径路由(评审#9): 命中 BOUNDARY 触发词且含预算词 → 走预算 playbook +BUDGET_TRIGGER_RE = re.compile(r"budget|预算", re.IGNORECASE) + +# 索要我方联系方式触发(15.2, doctrine v2): 客户问"你们怎么联系/你的微信多少" +# = 最高意向信号。提醒层按 source 分支: web 指按钮, App 无按钮走反向留资 +# - 排除第一人称将来时: "I'll call you later" 是日常告别语,不是索要联系方式。 +# 必须用定宽 negative lookbehind——lookahead 在动词位置求值,永远看不到 +# 前面的 let me / i'll,不生效。四个 lookbehind 全部定宽合法: i'll(5)、 +# i’ll(5, 弯撇 U+2019 手机常见)、i will(7)、let me(7);IGNORECASE 兼容大写 +CONTACT_ASK_TRIGGER_RE = re.compile( + r"(your|our)\s+(whatsapp|number|phone|email|contact)|" + r"(? int: + global _boundary_seq + _boundary_seq += 1 + return _boundary_seq + + +class ConversationManager: + def __init__(self, db: AsyncSession, ai_service: AIService | None = None): + self.db = db + self.ai = ai_service or AIService() + + async def create_conversation( + self, + source: ConversationSource = ConversationSource.standalone, + client_info: dict | None = None, + page_token: str | None = None, + context: dict | None = None, + ) -> Conversation: + session_id = uuid.uuid4().hex + conv = Conversation( + session_id=session_id, + source=source, + client_info=client_info, + page_token=page_token, + ) + self.db.add(conv) + await self.db.commit() + await self.db.refresh(conv) + + # Conditional whitelist seed of app context into raw_info (red line): + # only when `context` is provided, and only vehicle_name/vehicle_id + # mapped to app_* keys. WS path passes no context -> raw_info stays {}. + raw_info: dict = {} + if context: + if context.get("vehicle_name"): + raw_info["app_vehicle_name"] = context["vehicle_name"] + if context.get("vehicle_id"): + raw_info["app_vehicle_id"] = context["vehicle_id"] + profile = CustomerProfile(conversation_id=conv.id, raw_info=raw_info) + self.db.add(profile) + await self.db.commit() + + return conv + + async def get_by_session(self, session_id: str) -> Conversation | None: + result = await self.db.execute( + select(Conversation).where(Conversation.session_id == session_id) + ) + return result.scalar_one_or_none() + + async def get_by_page_token(self, page_token: str) -> Conversation | None: + result = await self.db.execute( + select(Conversation).where( + Conversation.page_token == page_token, + Conversation.status == ConversationStatus.active, + ) + ) + return result.scalar_one_or_none() + + async def stream_process_message( + self, session_id: str, content: str + ) -> AsyncGenerator[dict, None]: + """Stream AI reply token by token, then do extract/evaluate.""" + conv = await self.get_by_session(session_id) + if not conv: + yield {"type": "error", "message": "session not found"} + return + + if conv.status != ConversationStatus.active: + yield {"type": "error", "message": "conversation ended"} + return + + # Save user message + user_msg = Message( + conversation_id=conv.id, role=MessageRole.user, content=content + ) + self.db.add(user_msg) + await self.db.commit() + + history = await self._load_history(conv.id) + + # Build context reminder (merged into system prompt) + profile = await self._get_profile(conv.id) + + # P0 同轮竞态消除: 当前消息即时正则扫描——仅并入本轮内存视图 + # (提醒/话术路由),不落库;持久化仍由轮末 extractor+scanner 负责, + # 邮件触发时序不变。 + current_scan = scan_user_messages([{"role": "user", "content": content}]) + effective_raw = dict(profile.raw_info or {}) + if current_scan: + effective_raw["联系方式"] = merge_contacts( + current_scan, effective_raw.get("联系方式")) + + context_reminder = self._build_context_reminder( + effective_raw, content, history, source=conv.source.value) + if CONTACT_ASK_TRIGGER_RE.search(content or ""): + logger.info("Contact-ask reminder fired (source=%s)", conv.source.value) + + # Enrich short user messages with context so AI doesn't lose track + history_for_ai = list(history) + enriched = self._enrich_short_message(content, profile.raw_info or {}, history) + if history_for_ai and history_for_ai[-1]["role"] == "user" and enriched != content: + history_for_ai[-1] = {**history_for_ai[-1], "content": enriched} + logger.info("Enriched short message: %r -> %r", content[:50], enriched[:100]) + + # 方案B'(13.7): 句子粒度门控——干净句子逐句流出(恢复打字机), + # 违规句永不流出:已流出内容全部干净,违规时接过渡话术收尾并掐断生成。 + # F6(D6联动): 客户已留联系方式时,收尾话术不得再索要——路由 contact 版 + has_contact = bool(split_validate( + normalize_contacts(effective_raw.get("联系方式")))[0]) + + yield {"type": "stream_start"} + emitted = "" + pending = "" + violated = False + dropped_ask = False # Task 5b: 本轮是否有索要句被再索要墙丢弃 + dirty = "" + # 15.3: 双墙裁决——价格墙检查在前: 同句双违规时价格路径话术优先 + # (职能边界话术更具体,且自带留资引导);仅纯号码/邮箱违规时 + # violation=="no_contact",用零号码墙替换句 + violation = "" + async for token in self.ai.stream_chat(history_for_ai, context=context_reminder): + pending += token + complete, pending = split_sentences(pending) + for sentence in complete: + if violates_boundary(sentence): + violated = True + dirty = sentence + violation = "boundary" + break + if no_contact_value(sentence, history_for_ai): + violated = True + dirty = sentence + violation = "no_contact" + break + # Task 5b 再索要墙: 已留资时索要句静默丢弃(丢句不掐断, + # 其余句子照常流出——软于两墙的掐断+替换) + if has_contact and asks_contact(sentence): + logger.warning("Re-ask wall dropped sentence (conv=%s): %r", + conv.id, sentence[:60]) + dropped_ask = True + continue + for i in range(0, len(sentence), 40): + yield {"type": "stream_chunk", "content": sentence[i:i + 40]} + emitted += sentence + if violated: + break # 提前掐断生成,省 token + if not violated and pending: + if violates_boundary(pending): + violated = True + dirty = pending + violation = "boundary" + elif no_contact_value(pending, history_for_ai): + violated = True + dirty = pending + violation = "no_contact" + elif has_contact and asks_contact(pending): + # Task 5b 再索要墙: 尾巴是索要句 → 静默丢弃(不流出不掐断) + logger.warning("Re-ask wall dropped tail (conv=%s): %r", + conv.id, pending[:60]) + dropped_ask = True + else: + for i in range(0, len(pending), 40): + yield {"type": "stream_chunk", "content": pending[i:i + 40]} + emitted += pending + if violated: + if violation == "no_contact": + # 零号码墙替换句: 已留资→确认话术(不再索要);未留资按 source 分支 + # (App 无按钮 → 反向留资; web/widget/standalone → 指顶部按钮) + logger.warning( + "No-contact wall replaced sentence (conv=%s): %r", + conv.id, dirty[:60]) + closing = (NO_CONTACT_ALREADY if has_contact + else (NO_CONTACT_APP if conv.source.value == "app" + else NO_CONTACT_WEB)) + else: + logger.warning( + "Boundary filter intercepted sentence (conv=%s): %r | tail=%r", + conv.id, dirty[:80], (pending or "")[:40]) + closing = (pivot_for(content, has_contact) if emitted + else deflection_for(content, _boundary_rotate(), + has_contact=has_contact)) + for i in range(0, len(closing), 40): + yield {"type": "stream_chunk", "content": closing[i:i + 40]} + emitted += closing + # Task 5b 评审修正: 整条回复只有索要句(全被丢)→ 兜底流出"经理主动 + # 联系"句,绝不空回复。与上方墙收尾互斥——violated 路径已自带收尾, + # 此处仅在无任何实义内容流出时接管;已流出的纯空白前缀一并作废, + # 落库恒为完整兜底句(客户看到的比一个空格更糟的是: 什么都没有)。 + if not violated and dropped_ask and not emitted.strip(): + logger.warning("Re-ask wall dropped entire reply (conv=%s), " + "falling back to closing", conv.id) + for i in range(0, len(NO_CONTACT_ALREADY), 40): + yield {"type": "stream_chunk", + "content": NO_CONTACT_ALREADY[i:i + 40]} + emitted = NO_CONTACT_ALREADY + yield {"type": "stream_end"} + ai_reply = emitted # 落库与流出统一为已流出文本(下方持久化) + + # Save assistant message (skip if empty) + if ai_reply.strip(): + assistant_msg = Message( + conversation_id=conv.id, role=MessageRole.assistant, content=ai_reply + ) + self.db.add(assistant_msg) + await self.db.commit() + + # Round 2: Extract info (background) + full_history = history + [{"role": "assistant", "content": ai_reply}] + profile = await self._get_profile(conv.id) + + # Task 12b: 轮前联系方式快照(extractor 深合并 + 覆盖 + 同号同步的比对基线) + before_contacts = normalize_contacts((profile.raw_info or {}).get("联系方式")) + + new_info = await self.ai.extract_info( + full_history, existing_info=profile.raw_info or {} + ) + logger.info("Extractor result: %s", new_info) + if new_info: + raw = dict(profile.raw_info or {}) + for k, v in new_info.items(): + # 联系方式深合并(同key新值覆盖),其余顶层覆盖 —— 堵浅合并抹数据 + if k == "联系方式": + raw[k] = merge_contacts(raw.get(k), v) + else: + raw[k] = v + profile.raw_info = raw + self.db.add(profile) + + # Round 2.5: 联系方式正则辅源(S4: 扫全量客户消息历史) + # 🔴 合并方向: 扫描器只填空,extractor 已有值优先——否则扫描器的 + # 首个历史值会压掉 extractor 后续捕获的改号, D4 更正邮件失效 + scanned = scan_user_messages(full_history) + if scanned: + raw = dict(profile.raw_info or {}) + raw["联系方式"] = merge_contacts(scanned, raw.get("联系方式")) + profile.raw_info = raw + self.db.add(profile) + + # Task 12: 最新显式自述优先 —— 唯一允许文本压过 extractor 回传的位置 + # (extractor 受"保留已有信息"倾向会滞留旧值,实测 wrong number 场景命中) + latest_stated = latest_stated_contacts(full_history) + if latest_stated: + current = normalize_contacts((profile.raw_info or {}).get("联系方式")) + changed = False + for k, v in latest_stated.items(): + if v and canon(str(v)) != canon(str(current.get(k, ""))): + current[k] = v + changed = True + if changed: + raw = dict(profile.raw_info or {}) + raw["联系方式"] = current + profile.raw_info = raw + self.db.add(profile) + + # Task 12c 最近提及裁决: 镜像组分歧时以客户最晚提及的号码为准 + # (杀 extractor 历史振荡 + 压制幻觉;Task 12b 同号同步场景被涵盖) + after_contacts = normalize_contacts((profile.raw_info or {}).get("联系方式")) + resolved = resolve_by_recency(before_contacts, after_contacts, full_history) + if resolved != after_contacts: + raw = dict(profile.raw_info or {}) + raw["联系方式"] = resolved + profile.raw_info = raw + self.db.add(profile) + + # Round 3: Evaluate (仅 ≥3 条客户消息才有意义, 前两轮必然 pending —— D7) + user_msg_count = sum(1 for m in full_history if m["role"] == "user") + evaluation_ran = user_msg_count >= 3 + if evaluation_ran: + evaluation = await self.ai.evaluate( + profile=profile.raw_info or {}, + recent_messages=full_history, + ) + else: + # 客户端 evaluated 事件语义保持 pending;邮件标签由 evaluation_ran 决定(D2) + evaluation = {"intent_level": "pending", "reasoning": "", + "contact_info": {}, "info_sufficient": False} + logger.info("Evaluation (stream): profile=%s | eval=%s", + profile.raw_info, evaluation) + + # 邮件触发(D1): 与会话结束彻底解耦,联系方式到手即发 + pending_pushes: list[LeadPushLog] = [] + await self._maybe_send_lead_email( + conv, profile, evaluation, evaluated=evaluation_ran, + pending_pushes=pending_pushes, + ) + + # 联系方式判定(供 D8 评估落库): 有意向但没联系方式 → 卡单供 admin 人工兜底 + confirmed_now, _ = split_validate( + normalize_contacts((profile.raw_info or {}).get("联系方式"))) + + # 评估落库(D8 泛化): 每轮评估后刷新意向;email_sent/correction_count + # 永不触碰。有意向无联系方式 = 卡单(admin 可筛);已发过邮件的 + # contact_info 由触发器更正分支管理,这里不覆盖。 + if user_msg_count >= 3: + ev = await self._get_or_init_evaluation(conv.id) + ev.intent_level = IntentLevel(evaluation.get("intent_level", "pending")) + ev.reasoning = evaluation.get("reasoning", "") + if not confirmed_now: + ev.contact_info = {} + elif not ev.email_sent: + ev.contact_info = confirmed_now + self.db.add(ev) + + # 决策A(2026-08-18): 自动结束移除——会话永不因评估结果结束; + # evaluated 事件保留但 ended 恒为 False(客户端协议不变)。 + await self.db.commit() + + # 询盘推送(设计 8.1): 主事务提交后才派生后台任务——payload 内存 + # 传递,任务不读库只写结果;fire 为同步函数,纳秒级返回不阻塞 + lead_push_service.fire_pending_pushes(pending_pushes) + + yield { + "type": "evaluated", + "ended": False, + "intent_level": evaluation.get("intent_level", "pending"), + } + + async def _get_or_init_evaluation(self, conv_id: str) -> Evaluation: + result = await self.db.execute( + select(Evaluation).where(Evaluation.conversation_id == conv_id) + ) + ev = result.scalar_one_or_none() + if not ev: + ev = Evaluation( + conversation_id=conv_id, + intent_level=IntentLevel.pending, + contact_info={}, + ) + self.db.add(ev) + await self.db.flush() + return ev + + async def _recently_emailed_canons(self) -> set[str]: + cutoff = datetime.now(timezone.utc) - timedelta(hours=24) + # 只取需要的列;🔴 禁止加 LIMIT——完整 24h 窗口才正确 + result = await self.db.execute( + select(Evaluation.contact_info).where( + Evaluation.email_sent.is_(True), + Evaluation.evaluated_at >= cutoff, + ) + ) + seen: set[str] = set() + for (contact_info,) in result: + for v in (contact_info or {}).values(): + if v: + seen.add(canon(str(v))) + return seen + + async def _maybe_send_lead_email( + self, conv: Conversation, profile: CustomerProfile, evaluation: dict, + evaluated: bool = True, + pending_pushes: list[LeadPushLog] | None = None, + ) -> None: + """D1: 联系方式是发邮件的唯一条件。confirmed 才触发;suspect 只进提醒。 + + evaluated=False 表示本轮跳过了 GLM 评估(<3条客户消息)——此时邮件 + intent_level 必须传空串(→[未评估]),不得把客户端事件用的 "pending" + 透传成 [待确认](D2 两者语义不同)。 + """ + contacts_all = normalize_contacts((profile.raw_info or {}).get("联系方式")) + confirmed, _suspect = split_validate(contacts_all) + if not confirmed: + return + ev = await self._get_or_init_evaluation(conv.id) + + if not ev.email_sent: + # S3: 任一值已见于24h内其他会话 → 轻提醒;否则全量 + seen = await self._recently_emailed_canons() + if any(canon(str(v)) in seen for v in confirmed.values()): + await self._collect_push( + pending_pushes, ev, conv, profile, confirmed, + "returning", evaluation, evaluated) + sent = await email_service.send_returning_lead_email(confirmed) + else: + await self._collect_push( + pending_pushes, ev, conv, profile, confirmed, + "lead", evaluation, evaluated) + messages = await self._load_history(conv.id) + sent = await email_service.send_lead_email( + customer_profile=profile.raw_info or {}, + contact_info=confirmed, + reasoning=evaluation.get("reasoning", ""), + conversation_messages=messages, + intent_level=evaluation.get("intent_level", "") if evaluated else "", + ) + if sent: + ev.email_sent = True + ev.contact_info = confirmed + self.db.add(ev) + else: + old_canons = {canon(str(v)) for v in (ev.contact_info or {}).values()} + if any(canon(str(v)) not in old_canons for v in confirmed.values()): + if ev.correction_count < 3: + count = ev.correction_count + 1 + await self._collect_push( + pending_pushes, ev, conv, profile, confirmed, + "correction", evaluation, evaluated) + if await email_service.send_correction_email(confirmed, count=count): + ev.correction_count = count + ev.contact_info = confirmed + self.db.add(ev) + else: + # D4'(2026-08-18): 锁邮件不锁快照 —— 上限已到, + # admin 仍要看到客户最新号码 + ev.contact_info = confirmed + self.db.add(ev) + + async def _collect_push( + self, pending, ev, conv, profile, confirmed, + push_type, evaluation, evaluated, + ) -> None: + """询盘推送派发(设计 §8): 触发条件=进入发信分支,与 SMTP 成败解耦。 + + dispatch 内部吞异常(失败只记日志),这里无需再兜。 + """ + log = await lead_push_service.dispatch_lead_push( + self.db, ev, conv, profile, confirmed, push_type, + evaluation, evaluated, + ) + if log is not None and pending is not None: + pending.append(log) + + async def _load_history(self, conv_id: str) -> list[dict]: + result = await self.db.execute( + select(Message) + .where(Message.conversation_id == conv_id) + .order_by(Message.created_at) + ) + messages = result.scalars().all() + return [{"role": m.role.value, "content": m.content} for m in messages] + + async def _get_profile(self, conv_id: str) -> CustomerProfile: + result = await self.db.execute( + select(CustomerProfile).where(CustomerProfile.conversation_id == conv_id) + ) + return result.scalar_one() + + @staticmethod + def _enrich_short_message(content: str, raw_info: dict, history: list[dict] | None = None) -> str: + """Enrich short user messages with conversation context. + + Two sources: extractor output (primary), last substantive user + message (fallback when extractor hasn't run yet). + """ + if len(content.strip()) >= 30: + return content + + facts = [] + + # Primary: extractor output + if raw_info: + facts = [f"{k}: {v}" for k, v in raw_info.items() + if v and not str(k).startswith("_") and k != "联系方式"] + + # Fallback: use last substantive user message as context + if not facts and history: + for msg in reversed(history[:-1]): + if msg.get("role") == "user" and len(msg.get("content", "").strip()) >= 5: + facts.append("客户之前说: " + msg["content"][:80]) + break + + if not facts: + return content + + return ( + content + + "\n\n【系统上下文 - 系统标注,不要向客户提及】已知: " + + " | ".join(facts) + + "。不要重复询问这些已确认的信息!" + ) + + @staticmethod + def _build_context_reminder(raw_info: dict, latest_msg: str, + history: list[dict] | None = None, + source: str = "") -> str: + """Build a context reminder from extracted profile info.""" + # 职能边界提醒(评审#7: 前置到早退之前,首条消息画像为空也要生效) + boundary_note = "" + if BOUNDARY_TRIGGER_RE.search(latest_msg or ""): + if BUDGET_TRIGGER_RE.search(latest_msg or ""): + # 评审#9: 预算求推荐 → 预算路径(推荐+反问+引导),不一刀切要号 + boundary_note = ( + "⚠️ 职能边界提醒:客户报预算求推荐。推荐车型但绝不复述预算数字、" + "不报车价 → 反问意向车型 → 顺势引导留联系方式。") + else: + boundary_note = ( + "⚠️ 职能边界提醒:客户正在询问价格/优惠/交付时效/实质安排。绝对禁止" + "给出任何价格数字(包括复述客户自己的预算)、交付时间或任何承诺," + "按职能边界对应话术处理。") + + # 联系方式状态前移(供 boundary/contact 提醒共用判定) + contacts_all = normalize_contacts(raw_info.get("联系方式")) + confirmed, suspect = split_validate(contacts_all) + + # 索要我方联系方式提醒(15.2): 与 boundary_note 同样前移到早退之前, + # 空画像首条消息也要生效;按 source 分支(widget/standalone 有按钮, app 无)。 + # 冒烟实证: 抽象指令("用App话术")GLM 会无视并背 prompt 里更显眼的 web 脚本—— + # 提醒必须携带该分支的逐字话术,不让模型自己挑。 + contact_note = "" + if CONTACT_ASK_TRIGGER_RE.search(latest_msg or ""): + if source == "app": + if confirmed or suspect: + # 已留资: 索要我方联系方式 → 逐字"经理主动联系"版, + # 与"禁止再索要"提醒不再打架 + contact_note = ( + "⚠️ 客户正在索要我方联系方式,且客户已留过自己的联系方式。" + "请逐字使用这句:\"Our manager will reach out to you at your " + "WhatsApp shortly with the full quote!\"(中文同理)。" + "绝对禁止输出任何号码或邮箱,也绝对禁止再索要联系方式。") + else: + contact_note = ("⚠️ 客户正在索要我方联系方式(最高意向信号)。当前为 App 会话," + "没有页面按钮——禁止使用任何 tap/button 按钮话术!请逐字使用这句:" + "\"Drop your WhatsApp and our manager will reach out to you directly " + "with the full quote!\"(中文同理引导客户留联系方式)。" + "绝对禁止输出任何号码或邮箱。") + else: + contact_note = ("⚠️ 客户正在索要我方联系方式(最高意向信号)。请逐字使用这句指路:" + "\"Just tap the WhatsApp button at the top of this chat — " + "our manager replies fast!\"(邮箱同理指 Email 按钮)。" + "绝对禁止口头报出任何号码或邮箱。") + + if not raw_info: + # 空画像:仅返回边界/联系提醒(可能为空串) + return "\n".join(x for x in (boundary_note, contact_note) if x) + + parts = ["[上下文提醒 - 这是一条系统指令,绝对不要向客户提及这段内容]"] + parts.append("客户已明确告知以下信息,你绝对不能再问这些:") + + for key, val in raw_info.items(): + if str(key).startswith("_") or key == "联系方式": + continue # 内部状态与联系方式单独渲染 + if val: + parts.append(f"{key}: {val}") + + # 联系方式专项(D6) 与 职能边界(D6联动) + if confirmed: + shown = ", ".join( + f"{k}: {' '.join(str(v).split())[:60]}" for k, v in confirmed.items()) + # 逐字话术直接引用 NO_CONTACT_ALREADY 常量(Task 5 已 import)—— + # 与零号码墙替换句同源,三处"经理主动联系"话术不漂移 + parts.append(f"⚠️ 客户已提供联系方式({shown})。绝对禁止再索要任何联系方式!" + f"需要回应时可直接逐字使用:\"{NO_CONTACT_ALREADY}\"(中文同理)。" + f"确认收到后继续解答客户问题即可,不必急着结束对话。") + if boundary_note: + parts.append(boundary_note) + parts.append("客户已留过联系方式——边界请求直接告知'客户经理会把准确报价/" + "安排发到你 WhatsApp'即可,禁止再索要联系方式。") + elif suspect: + shown = ", ".join( + f"{k}: {' '.join(str(v).split())[:60]}" for k, v in suspect.items()) + parts.append(f"⚠️ 客户疑似已提供联系方式({shown})。禁止重新索要;" + f"可礼貌地请客户确认一次。") + if boundary_note: + parts.append(boundary_note) + + if boundary_note and not confirmed and not suspect: + parts.append(boundary_note) # 未留资:边界提醒引导留资(职能边界话术自带) + + # 15.2: 索要我方联系方式提醒注入(镜像 boundary_note 的挂载位)。 + # 互留提示仅限 web 来源且客户未留资(confirmed/suspect 都算"已留"); + # App 的反向留资话术本身就是主话术,不再叠加 + if contact_note: + note = contact_note + if source != "app" and not confirmed and not suspect: + note += "客户尚未留过联系方式,可顺势轻问一句(每会话最多一次)。" + parts.append(note) + + if len(latest_msg.strip()) < 25: + parts.append("⚠️ 客户发的是短消息!必须结合上面已知信息来理解,绝对不能当作独立问题!") + + return "\n".join(parts) diff --git a/app/services/email_service.py b/app/services/email_service.py new file mode 100644 index 0000000..f7ee35d --- /dev/null +++ b/app/services/email_service.py @@ -0,0 +1,129 @@ +import html +import logging +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText + +import aiosmtplib + +from app.config import settings + +logger = logging.getLogger(__name__) + +TAG_LABELS = { + "interested": "意向强", "pending": "待确认", + "not_interested": "无意向", "": "未评估", +} +CONTACT_LABELS = {"phone": "手机", "whatsapp": "WhatsApp", "email": "邮箱", "wechat": "微信"} + + +def _esc(v) -> str: + return html.escape(str(v)) if v else "" + + +def _flat(v) -> str: + """主题行用:折叠空白(堵换行注入头部的通道)+ 限长。""" + return " ".join(str(v).split())[:60] + + +def _contact_rows(contacts: dict) -> str: + return "".join( + f"
{_esc(reasoning)}
+注:邮件为触发时快照,对话可能仍在进行,最新记录请以后台为准。
+ + """ + msg.attach(MIMEText(html_body, "html", "utf-8")) + return await _send(msg) + + +async def send_returning_lead_email(contact_info: dict) -> bool: + """D5: 24h 内同联系方式再访,只发一行轻提醒。""" + contact_parts = [ + f"{CONTACT_LABELS.get(k, k)}: {_flat(v)}" for k, v in (contact_info or {}).items() if v + ] + msg = MIMEMultipart("alternative") + msg["Subject"] = f"[线索回访] {' | '.join(contact_parts) or '已知客户'}再次来访" + msg["From"] = f"【XYH】 <{settings.SMTP_USER}>" + msg["To"] = settings.NOTIFY_EMAIL + msg.attach(MIMEText( + f"该联系方式 24h 内已发过完整询盘,客户又开启了新对话," + f"详情请查后台。
", "html", "utf-8")) + return await _send(msg) + + +async def send_correction_email(contact_info: dict, count: int = 1) -> bool: + """D4: 联系方式变化(改号/新增/无提示换值)后的轻量更正邮件,每会话上限3封。""" + msg = MIMEMultipart("alternative") + msg["Subject"] = f"[询盘·更正{count}] 客户联系方式已更新(以本封为准)" + msg["From"] = f"【XYH】 <{settings.SMTP_USER}>" + msg["To"] = settings.NOTIFY_EMAIL + msg.attach(MIMEText( + f"这是本会话第 {count} 次更正(上限 3 次),请以本封为准:
" + f"