tw/core/agent_pre_rules.py

from __future__ import annotations

import logging
import random
from datetime import datetime
from typing import TYPE_CHECKING, Optional

from core.rules import Rule, RuleContext, RuleEngine, RuleResult
from services.risk_service import RiskService

if TYPE_CHECKING:
    from core.pydantic_ai_agent import (
        AgentResponse,
        ConversationState,
        CustomerMessage,
        CustomerServiceAgent,
    )


class AgentPreRuleService:
    """Pre-processing rule chain for short replies, cooldown, and text risk."""

    def __init__(self, agent: "CustomerServiceAgent", risk_service: RiskService):
        self.agent = agent
        self.risk_service = risk_service
        self.engine = self._build_engine()

    async def run(
        self,
        *,
        message: "CustomerMessage",
        state: "ConversationState",
        trace_id: str,
    ) -> Optional["AgentResponse"]:
        ctx = RuleContext(data={"message": message, "state": state, "trace_id": trace_id})
        result = await self.engine.run(ctx)
        if not result.stop:
            return None
        response = result.payload.get("response")
        return response

    def _build_engine(self) -> RuleEngine:
        return RuleEngine(
            rules=[
                Rule(
                    name="meaningless_short_text",
                    priority=10,
                    predicate=self._rule_pred_meaningless_short_text,
                    action=self._rule_act_meaningless_short_text,
                ),
                Rule(
                    name="cooldown_silent",
                    priority=20,
                    predicate=self._rule_pred_cooldown_silent,
                    action=self._rule_act_cooldown_silent,
                ),
                Rule(
                    name="manual_risk_block",
                    priority=30,
                    predicate=self._rule_pred_manual_risk_block,
                    action=self._rule_act_manual_risk_block,
                ),
                Rule(
                    name="text_risk_block",
                    priority=40,
                    predicate=self._rule_pred_text_risk_block,
                    action=self._rule_act_text_risk_block,
                ),
            ]
        )

    async def _rule_pred_meaningless_short_text(self, ctx: RuleContext) -> bool:
        message = ctx.get("message")
        state = ctx.get("state")
        return self.agent._should_handle_as_meaningless_short_text(state, message.msg)

    async def _rule_act_meaningless_short_text(self, ctx: RuleContext) -> RuleResult:
        from core.pydantic_ai_agent import AgentResponse

        message = ctx.get("message")
        state = ctx.get("state")
        trace_id = ctx.get("trace_id", "")
        ping = random.choice(("嗯咯", "嗯啦", "嗯", "哦"))
        state.last_reply_at = datetime.now()
        self.agent._activity_log(
            "agent_ping_reply",
            trace_id=trace_id,
            customer_id=message.from_id,
            msg=message.msg,
            reply=ping,
        )
        return RuleResult(
            matched=True,
            stop=True,
            action="agent_ping_reply",
            payload={"response": AgentResponse(reply=ping, should_reply=True, need_transfer=False)},
        )

    async def _rule_pred_cooldown_silent(self, ctx: RuleContext) -> bool:
        message = ctx.get("message")
        state = ctx.get("state")
        return self.agent._in_cooldown(state, message.msg)

    async def _rule_act_cooldown_silent(self, ctx: RuleContext) -> RuleResult:
        from core.pydantic_ai_agent import AgentResponse

        message = ctx.get("message")
        state = ctx.get("state")
        trace_id = ctx.get("trace_id", "")
        elapsed = int((datetime.now() - state.last_reply_at).total_seconds()) if state.last_reply_at else 0
        logger.info("[Agent] 冷却期静默（距上次回复 %ss）：%r", elapsed, message.msg)
        self.agent._activity_log(
            "agent_cooldown_silent",
            trace_id=trace_id,
            customer_id=message.from_id,
            elapsed_s=elapsed,
        )
        return RuleResult(
            matched=True,
            stop=True,
            action="agent_cooldown_silent",
            payload={"response": AgentResponse(reply="", should_reply=False, need_transfer=False)},
        )

    async def _rule_pred_manual_risk_block(self, ctx: RuleContext) -> bool:
        message = ctx.get("message")
        decision = self.risk_service.check_manual_block(message.from_id)
        ctx.set("manual_risk_decision", decision)
        return decision.blocked

    async def _rule_act_manual_risk_block(self, ctx: RuleContext) -> RuleResult:
        from core.pydantic_ai_agent import AgentResponse, TRANSFER_MESSAGE

        message = ctx.get("message")
        trace_id = ctx.get("trace_id", "")
        decision = ctx.get("manual_risk_decision")
        self.agent._activity_log(
            "agent_manual_risk_reject",
            trace_id=trace_id,
            customer_id=message.from_id,
            risk=(decision.profile if decision else {}),
        )
        return RuleResult(
            matched=True,
            stop=True,
            action="agent_manual_risk_reject",
            payload={
                "response": AgentResponse(
                    reply="这边无法继续为你处理该类需求，给你转人工专员对接。",
                    should_reply=True,
                    need_transfer=True,
                    transfer_msg=TRANSFER_MESSAGE,
                )
            },
        )

    async def _rule_pred_text_risk_block(self, ctx: RuleContext) -> bool:
        message = ctx.get("message")
        decision = await self.risk_service.check_text_block(
            message.msg,
            political_detector=self.agent._is_political_inquiry,
            map_detector=self.agent._is_map_inquiry,
        )
        ctx.set("text_risk_decision", decision)
        return decision.blocked

    async def _rule_act_text_risk_block(self, ctx: RuleContext) -> RuleResult:
        from core.pydantic_ai_agent import AgentResponse

        message = ctx.get("message")
        state = ctx.get("state")
        trace_id = ctx.get("trace_id", "")
        decision = ctx.get("text_risk_decision")
        state.pending_image_urls.clear()
        state.pending_requirements.clear()
        self.agent._sync_pending_quote_state(message.from_id, state)

        reject_text = self.risk_service.build_reject_text(decision.category if decision else "other")
        reply = await self.agent._rewrite_reply_with_ai(
            message=message,
            state=state,
            reply=reject_text,
            scene="risk_reject",
        )
        state.last_reply_at = datetime.now()
        logger.info("[REPLY->CUSTOMER] %s", reply)
        self.agent._activity_log(
            "agent_risk_reject",
            trace_id=trace_id,
            customer_id=message.from_id,
            risk_category=(decision.category if decision else "other"),
            risk_source=(decision.source if decision else "unknown"),
            reply=reply,
        )
        return RuleResult(
            matched=True,
            stop=True,
            action="agent_risk_reject",
            payload={"response": AgentResponse(reply=reply, should_reply=True, need_transfer=False)},
        )
logger = logging.getLogger("cs_agent")