feat: add governed customer service agent

.env.example

@@ -38,6 +38,7 @@ MODEL_DEFAULT_ENDPOINT=
 MODEL_FALLBACK_ENDPOINT=
 KNOWLEDGE_EMBEDDING_ENDPOINT_CODE=
 KNOWLEDGE_EMBEDDING_TIMEOUT_MS=15000
+CUSTOMER_SERVICE_PHONE=15936583816
 
 SSE_HEARTBEAT_SECONDS=15
 SSE_MAX_CONNECTION_SECONDS=300

app/core/config.py

@@ -24,6 +24,7 @@ class Settings(BaseSettings):
     jwt_public_key_path: str = "config/jwt/jwt-public.pem"
     jwt_clock_skew_seconds: int = Field(default=30, ge=0)
     visitor_token_ttl_seconds: int = Field(default=900, ge=60, le=3600)
+    customer_service_phone: str = Field(default="15936583816", pattern=r"^1\d{10}$")
     mysql_dsn: str
     mysql_pool_size: int = Field(default=5, ge=1)
     mysql_max_overflow: int = Field(default=10, ge=0)

app/core/contracts.py

@@ -11,6 +11,8 @@ class AgentRequestMetadata(BaseModel):
     locale: str | None = None
     client_version: str | None = None
     ui_entry: str | None = None
+    # 仅由受理服务写入 Outbox，客户端 API 不接收该字段。
+    chitchat_streak: int = Field(default=0, ge=0, le=5)
 
 
 class AgentRequest(BaseModel):

app/main.py

@@ -1,5 +1,8 @@
+from pathlib import Path
+
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
+from fastapi.staticfiles import StaticFiles
 
 from app.api.controllers.admin import router as admin_router
 from app.api.controllers.agent_runs import router as agent_runs_router
@@ -52,6 +55,11 @@ def create_app() -> FastAPI:
     application.include_router(knowledge_router)
     application.include_router(health_router)
     application.include_router(admin_router)
+    application.mount(
+        "/customer-service-test",
+        StaticFiles(directory=Path(__file__).resolve().parent / "static", html=True),
+        name="customer-service-test",
+    )
     return application
 
 

app/service/agent/bootstrap.py

@@ -12,6 +12,7 @@ from app.core.nl2sql_contracts import FinancialNL2SQLInput
 from app.infrastructure.fund_quote_cache import FundQuoteCache
 from app.infrastructure.memory_cache import MemoryCacheAdapter
 from app.infrastructure.vector_memory import VectorMemoryAdapter
+from app.service.agent.customer_service_agent import CustomerServiceAgent
 from app.service.agent.factory import AgentFactory
 from app.service.agent.governance import PlatformGovernance
 from app.service.agent.implementations.fund_query_demo import FundQueryDemoAgent
@@ -198,3 +199,7 @@ def register_business_agents(factory: AgentFactory) -> None:
         OffsiteFundAgent.definition,
         lambda _context: OffsiteFundAgent(OffsiteFundAgent.definition),
     )
+    factory.register(
+        CustomerServiceAgent.definition,
+        lambda _context: CustomerServiceAgent(),
+    )

app/service/agent/customer_service_agent.py

@@ -0,0 +1,113 @@
+"""奶龙基金智能助手：只处理一期公开客服范围。"""
+
+from app.core.config import get_settings
+from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext
+from app.core.errors import RecoverableAgentError
+from app.core.knowledge_contracts import KnowledgeSearchResult
+from app.service.agent.base import BaseAgent
+from app.service.agent.customer_service_routing import CustomerServiceIntentRouter
+
+
+def _human_service_phone() -> str:
+    return get_settings().customer_service_phone
+
+
+class CustomerServiceAgent(BaseAgent):
+    definition = AgentDefinition(
+        agent_type="customer_service",
+        version="1.0.0",
+        allowed_tools=("query_knowledge",),
+        allowed_roles=("visitor", "customer"),
+        allowed_portals=("api",),
+        requires_model_intent_classification=False,
+        supported_intents=(
+            "security_notice",
+            "account_entry",
+            "human_transfer",
+            "compliance_refusal",
+            "chitchat",
+            "public_knowledge",
+            "faq",
+            "product_inquiry",
+            "policy_explain",
+        ),
+    )
+
+    def __init__(self) -> None:
+        super().__init__(self.definition)
+
+    async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
+        message = request.message.strip()
+        route = CustomerServiceIntentRouter.classify(message)
+        if route.intent == "security_notice":
+            return CoreResult(
+                text=(
+                    "请立即停止操作，不要继续提供验证码、密码或身份证件信息，并尽快联系"
+                    f"人工客服 {_human_service_phone()}。"
+                ),
+                transfer_required=True,
+                transfer_reason="security_notice",
+            )
+        if route.intent == "compliance_refusal":
+            return CoreResult(
+                text=(
+                    "我不能推荐具体产品、承诺收益或代您交易；如需进一步了解，请联系"
+                    f"人工客服 {_human_service_phone()}。"
+                ),
+                transfer_required=True,
+                transfer_reason="compliance_refusal",
+            )
+        if route.intent == "account_entry":
+            if "visitor" in context.roles:
+                return CoreResult(text="我无法查询账户数据，请先登录后前往“我的账户”查看相关状态。")
+            return CoreResult(text="我无法查询账户数据，请前往“我的账户”查看相关状态。")
+        if route.intent == "human_transfer":
+            return CoreResult(
+                text=(
+                    f"您可以联系人工客服 {_human_service_phone()}（工作日 09:00-18:00）"
+                    "进一步核实。"
+                ),
+                transfer_required=True,
+                transfer_reason="human_transfer",
+            )
+        if route.is_chitchat:
+            if request.metadata.chitchat_streak == 4:
+                return CoreResult(text="和您聊天很开心呀。您是想了解相关的基金业务或公开信息吗？")
+            return CoreResult(text="您好呀，我是奶龙基金智能助手，很高兴和您聊天。")
+        try:
+            output = await self.call_tool(
+                "query_knowledge",
+                {"query": message, "intents": route.knowledge_intents, "top_k": 5},
+                intent="public_knowledge",
+                context=context,
+            )
+            result = (
+                output
+                if isinstance(output, KnowledgeSearchResult)
+                else KnowledgeSearchResult.model_validate(output)
+            )
+        except RecoverableAgentError:
+            return self._knowledge_transfer("knowledge_unavailable")
+        answers = [
+            hit.answer.strip()
+            for hit in result.hits
+            if isinstance(hit.answer, str) and hit.answer.strip()
+        ]
+        if not answers:
+            return self._knowledge_transfer("knowledge_not_found")
+        return CoreResult(text=answers[0])
+
+    @staticmethod
+    def _knowledge_transfer(reason: str) -> CoreResult:
+        phone = _human_service_phone()
+        if reason == "knowledge_unavailable":
+            text = (
+                "您好呀，我暂时无法从公开资料中确认这个问题，请联系"
+                f"人工客服 {phone} 进一步核实。"
+            )
+        else:
+            text = (
+                "您好呀，我暂时没有查到匹配的公开资料，请联系"
+                f"人工客服 {phone} 进一步核实。"
+            )
+        return CoreResult(text=text, transfer_required=True, transfer_reason=reason)

app/service/agent/customer_service_routing.py

@@ -0,0 +1,74 @@
+"""一期客服的确定性路由，先处理安全和边界，再允许公开知识检索。"""
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class CustomerServiceRoute:
+    intent: str
+    knowledge_intents: tuple[str, ...] = ()
+    is_chitchat: bool = False
+
+
+class CustomerServiceIntentRouter:
+    _SECURITY_KEYWORDS = ("验证码", "密码泄露", "被盗", "诈骗", "非本人交易")
+    _COMPLIANCE_KEYWORDS = ("推荐", "收益最高", "稳赚", "保本", "帮我买", "替我交易")
+    _ACCOUNT_KEYWORDS = ("持仓", "收益", "订单", "定投", "银行卡", "风险测评", "投诉进度")
+    _HUMAN_TRANSFER_KEYWORDS = ("转人工", "人工客服", "投诉", "赔偿", "法律", "纠纷")
+    _POLICY_KEYWORDS = (
+        "申购", "赎回", "到账", "费率", "手续费", "确认份额", "交易日", "分红", "规则", "政策"
+    )
+    _PRODUCT_KEYWORDS = (
+        "产品", "基金代码", "基金经理", "份额类别", "a类", "c类", "净值", "风险等级"
+    )
+    _CHITCHAT_MESSAGES = frozenset({
+        "你好", "您好", "嗨", "哈喽", "在吗", "谢谢", "谢谢你", "再见", "拜拜",
+        "你是谁", "你叫什么", "你今天开心吗",
+    })
+    _CHITCHAT_PHRASES = ("今天天气", "讲个笑话", "你几岁", "你开心吗", "你忙吗")
+
+    @classmethod
+    def classify(cls, message: str) -> CustomerServiceRoute:
+        normalized = message.strip().lower()
+        if cls._contains(normalized, cls._SECURITY_KEYWORDS):
+            return CustomerServiceRoute(intent="security_notice")
+        if cls._contains(normalized, cls._COMPLIANCE_KEYWORDS):
+            return CustomerServiceRoute(intent="compliance_refusal")
+        if cls._contains(normalized, cls._ACCOUNT_KEYWORDS):
+            return CustomerServiceRoute(intent="account_entry")
+        if cls._contains(normalized, cls._HUMAN_TRANSFER_KEYWORDS):
+            return CustomerServiceRoute(intent="human_transfer")
+        if cls._is_chitchat(normalized):
+            return CustomerServiceRoute(intent="chitchat", is_chitchat=True)
+        if cls._contains(normalized, cls._POLICY_KEYWORDS):
+            return CustomerServiceRoute(
+                intent="public_knowledge", knowledge_intents=("policy_explain",)
+            )
+        if cls._contains(normalized, cls._PRODUCT_KEYWORDS):
+            return CustomerServiceRoute(
+                intent="public_knowledge", knowledge_intents=("product_inquiry",)
+            )
+        return CustomerServiceRoute(intent="public_knowledge", knowledge_intents=("faq",))
+
+    @classmethod
+    def chitchat_streak(cls, prior_messages: Sequence[str], message: str) -> int:
+        """返回当前消息在同一会话中连续闲聊的次数，最大只需记录到第五句。"""
+        if not cls._is_chitchat(message.strip().lower()):
+            return 0
+        streak = 1
+        for prior_message in reversed(prior_messages):
+            if not cls._is_chitchat(prior_message.strip().lower()):
+                break
+            streak += 1
+            if streak == 5:
+                break
+        return streak
+
+    @staticmethod
+    def _contains(message: str, keywords: tuple[str, ...]) -> bool:
+        return any(keyword in message for keyword in keywords)
+
+    @classmethod
+    def _is_chitchat(cls, message: str) -> bool:
+        return message in cls._CHITCHAT_MESSAGES or cls._contains(message, cls._CHITCHAT_PHRASES)

app/service/agent/governance.py

@@ -6,6 +6,7 @@ from typing import Protocol
 from sqlalchemy import select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.config import get_settings
 from app.core.contracts import (
     AgentDefinition,
     AgentResult,
@@ -135,8 +136,13 @@ def review_output(
     else:
         # Apply to text and citation titles, not only to the displayed answer.
         def redact(value: str) -> str:
+            # 人工客服电话是唯一经过配置审核、可在对客文本中保留的号码。
+            trusted_phone = get_settings().customer_service_phone
+            protected_phone = "__customer_service_phone__"
+            value = value.replace(trusted_phone, protected_phone)
             value = re.sub(r"(?<!\d)1[3-9]\d{9}(?!\d)", "[手机号已脱敏]", value)
-            return re.sub(r"(?<!\d)\d{15,19}[Xx]?(?!\d)", "[敏感号码已脱敏]", value)
+            value = re.sub(r"(?<!\d)\d{15,19}[Xx]?(?!\d)", "[敏感号码已脱敏]", value)
+            return value.replace(protected_phone, trusted_phone)
 
         content = content.model_copy(update={
             "text": redact(output),

app/service/agent_persistence_service.py

@@ -9,7 +9,7 @@ from app.core.contracts import AgentResult, DomainEvent
 from app.core.errors import RunLeaseLostError
 from app.model.audit import InteractionAudit
 from app.model.conversation import ConversationMessage
-from app.model.platform import AgentRun, DomainEventOutbox, RequestIdempotency
+from app.model.platform import AgentRun, DomainEventOutbox, HandoverTicket, RequestIdempotency
 
 
 class AgentPersistenceService:
@@ -52,6 +52,31 @@ class AgentPersistenceService:
             )
             self.session.add(message)
             await self.session.flush()
+            # 转人工只接受已完成治理决策的结果；工单与回复绑定，便于管理员回看上下文。
+            handover_ticket: HandoverTicket | None = None
+            if result.result.transfer_required:
+                handover_ticket = HandoverTicket(
+                    ticket_no=f"ticket-{uuid4().hex[:24]}",
+                    session_id=run.session_id,
+                    customer_id=run.user_id,
+                    source_agent=run.agent_type,
+                    source_message_id=message.id,
+                    intent=(result.result.intent.intent if result.result.intent else None),
+                    confidence=(
+                        Decimal(str(result.result.intent.confidence))
+                        if result.result.intent else None
+                    ),
+                    reason_code=result.result.transfer_reason or "agent_requested",
+                    conversation_summary=result.result.text,
+                    source_references=[
+                        reference.model_dump(mode="json")
+                        for reference in result.result.source_references
+                    ],
+                    status="pending",
+                    created_at=now,
+                    updated_at=now,
+                )
+                self.session.add(handover_ticket)
             run.result_message_id = message.id
             run.status = "succeeded"
             run.result_version = 1
@@ -63,6 +88,19 @@ class AgentPersistenceService:
                 session_id=run.session_id, portal="agent", action_type="agent.run_completed",
                 detail={"run_id": run_id, "result_message_id": message.id}, created_at=now,
             ))
+            if handover_ticket is not None:
+                # 管理员工单列表之外还需留一条不可变审计，记录 Agent 自动转接来源。
+                self.session.add(InteractionAudit(
+                    actor_type="agent", actor_id=run.user_id, target_customer_id=run.user_id,
+                    session_id=run.session_id, portal="agent",
+                    action_type="agent.handover_requested",
+                    detail={
+                        "run_id": run_id,
+                        "ticket_no": handover_ticket.ticket_no,
+                        "reason_code": handover_ticket.reason_code,
+                    },
+                    created_at=now,
+                ))
             await self.session.execute(
                 update(RequestIdempotency)
                 .where(RequestIdempotency.id == run.idempotency_id)
@@ -80,6 +118,14 @@ class AgentPersistenceService:
                     payload={"run_id": run_id, "message_id": message.id,
                              "customer_id": run.user_id}, occurred_at=now,
                 ))
+            if handover_ticket is not None:
+                # Outbox 事件由后续管理员通知/工单消费方可靠投递，服务层不直接通知外部系统。
+                events.append(DomainEvent(
+                    event_id=str(uuid4()), event_type="conversation.transfer_requested",
+                    aggregate_type="conversation", aggregate_id=run.session_id,
+                    trace_id=run.trace_id,
+                    payload={"ticket_no": handover_ticket.ticket_no}, occurred_at=now,
+                ))
             for event in events:
                 self.session.add(DomainEventOutbox(
                     event_id=event.event_id, event_type=event.event_type,

app/service/agent_run_application_service.py

@@ -1,5 +1,6 @@
 import hashlib
 import json
+from collections.abc import Sequence
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from uuid import uuid4
@@ -20,6 +21,7 @@ from app.model.platform import AgentRun, RequestIdempotency
 from app.model.session import ConversationSession
 from app.repository.outbox_repository import OutboxRepository
 from app.service.agent.bootstrap import get_agent_factory
+from app.service.agent.customer_service_routing import CustomerServiceIntentRouter
 from app.service.agent.factory import AgentFactory
 
 
@@ -30,6 +32,20 @@ class RunAccepted:
     status: str = "queued"
 
 
+def build_outbox_metadata(
+    request: AgentRequest, prior_user_messages: Sequence[str]
+) -> dict[str, object]:
+    """构造 Worker 使用的内部元数据，不信任外部传入的闲聊计数。"""
+    metadata = request.metadata
+    if request.agent_type == "customer_service":
+        metadata = metadata.model_copy(update={
+            "chitchat_streak": CustomerServiceIntentRouter.chitchat_streak(
+                prior_user_messages, request.message
+            )
+        })
+    return metadata.model_dump(mode="json")
+
+
 class AgentRunApplicationService:
     def __init__(self, session: AsyncSession, factory: AgentFactory | None = None) -> None:
         self.session = session
@@ -89,6 +105,20 @@ class AgentRunApplicationService:
                     raise RuntimeError("idempotency record has no run")
                 return RunAccepted(run.run_id, run.trace_id, run.status)
 
+            # 仅读取当前会话最近三条用户消息；第四条闲聊即触发一次自然业务引导。
+            prior_user_messages = list(await self.session.scalars(
+                select(ConversationMessage.content)
+                .where(
+                    ConversationMessage.session_id == request.session_id,
+                    ConversationMessage.role == "user",
+                )
+                .order_by(ConversationMessage.id.desc())
+                .limit(3)
+            ))
+            outbox_metadata = build_outbox_metadata(
+                request, tuple(reversed(prior_user_messages))
+            )
+
             trace_id = context.trace_id
             message = ConversationMessage(
                 session_id=request.session_id, customer_id=user_id, portal="api",
@@ -121,7 +151,7 @@ class AgentRunApplicationService:
                 payload={
                     "run_id": run_id,
                     "actor_type": "visitor" if "visitor" in context.roles else "authenticated",
-                    "metadata": request.metadata.model_dump(mode="json"),
+                    "metadata": outbox_metadata,
                 },
                 occurred_at=now,
             ))

app/static/index.html

@@ -0,0 +1,155 @@
+<!doctype html>
+<html lang="zh-CN">
+<head>
+  <meta charset="UTF-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>奶龙基金智能助手联调</title>
+  <style>
+    :root { --ink: #17212b; --muted: #607080; --line: #d9e1e7; --canvas: #f4f7f8; --surface: #ffffff; --brand: #008a7a; --brand-dark: #00695f; --accent: #e96f45; --visitor: #e8f6f1; --agent: #f2f5f7; --danger: #b43b2c; }
+    * { box-sizing: border-box; }
+    body { margin: 0; min-height: 100vh; background: var(--canvas); color: var(--ink); font-family: "Microsoft YaHei", "PingFang SC", sans-serif; }
+    main { width: min(100%, 980px); min-height: 100vh; margin: 0 auto; padding: 24px; display: grid; grid-template-rows: auto auto minmax(360px, 1fr) auto; gap: 14px; }
+    header { display: flex; align-items: center; justify-content: space-between; gap: 16px; padding-bottom: 14px; border-bottom: 1px solid var(--line); }
+    .brand { display: flex; align-items: center; gap: 12px; }
+    .brand-mark { width: 38px; height: 38px; display: grid; place-items: center; background: var(--brand); color: #ffffff; border-radius: 8px; font-weight: 700; }
+    h1 { margin: 0; font-size: 20px; font-weight: 700; letter-spacing: 0; }
+    .status { margin: 2px 0 0; color: var(--muted); font-size: 13px; }
+    .role { color: var(--brand-dark); font-weight: 700; }
+    .samples { display: flex; flex-wrap: wrap; gap: 8px; }
+    .sample { min-height: 34px; border: 1px solid var(--line); background: var(--surface); color: var(--ink); padding: 7px 10px; border-radius: 6px; cursor: pointer; font: inherit; font-size: 13px; }
+    .sample:hover, .sample:focus-visible { border-color: var(--brand); color: var(--brand-dark); outline: none; }
+    .chat { overflow-y: auto; display: flex; flex-direction: column; gap: 12px; padding: 18px; background: var(--surface); border: 1px solid var(--line); border-radius: 8px; }
+    .message { display: flex; gap: 10px; max-width: min(82%, 620px); }
+    .message.user { align-self: flex-end; flex-direction: row-reverse; }
+    .avatar { flex: 0 0 30px; height: 30px; display: grid; place-items: center; border-radius: 50%; background: var(--brand); color: #ffffff; font-size: 12px; font-weight: 700; }
+    .message.user .avatar { background: var(--accent); }
+    .bubble { margin: 0; padding: 10px 12px; background: var(--agent); border-radius: 6px; font-size: 14px; line-height: 1.6; white-space: pre-wrap; overflow-wrap: anywhere; }
+    .message.user .bubble { background: var(--visitor); }
+    .pending .bubble { color: var(--muted); }
+    .error .bubble { background: #fff0ed; color: var(--danger); }
+    form { display: grid; grid-template-columns: minmax(0, 1fr) auto; gap: 10px; }
+    input { min-width: 0; height: 44px; border: 1px solid var(--line); border-radius: 6px; padding: 0 12px; background: var(--surface); color: var(--ink); font: inherit; }
+    input:focus { border-color: var(--brand); outline: 2px solid #bde6df; }
+    button[type="submit"] { min-width: 76px; height: 44px; border: 0; border-radius: 6px; background: var(--brand); color: #ffffff; cursor: pointer; font: inherit; }
+    button[type="submit"]:disabled { cursor: wait; opacity: 0.6; }
+    @media (max-width: 620px) { main { padding: 14px; } header { align-items: flex-start; flex-direction: column; } .message { max-width: 92%; } }
+  </style>
+</head>
+<body>
+  <main>
+    <header>
+      <div class="brand">
+        <div class="brand-mark" aria-hidden="true">龙</div>
+        <div><h1>奶龙基金智能助手</h1><p class="status">公开知识与客服分流联调页</p></div>
+      </div>
+      <p id="connection-status" class="status"><span class="role">访客</span>：正在建立测试连接</p>
+    </header>
+    <section class="samples" aria-label="测试场景">
+      <button class="sample" type="button" data-message="你好">闲聊</button>
+      <button class="sample" type="button" data-message="基金赎回到账规则是什么">政策</button>
+      <button class="sample" type="button" data-message="我的持仓收益是多少">账户入口</button>
+      <button class="sample" type="button" data-message="验证码已经发给别人了">安全风险</button>
+      <button class="sample" type="button" data-message="帮我推荐一只收益最高的基金">合规拒答</button>
+      <button class="sample" type="button" data-message="我要转人工客服">人工服务</button>
+    </section>
+    <section id="chat" class="chat" aria-live="polite"></section>
+    <form id="composer">
+      <input id="message" type="text" maxlength="2000" autocomplete="off" placeholder="输入要测试的问题" aria-label="客服测试消息">
+      <button id="send" type="submit">发送</button>
+    </form>
+  </main>
+  <script>
+    let accessToken = "";
+    const sessionId = `visitor-${crypto.randomUUID()}`;
+    const chat = document.querySelector("#chat");
+    const input = document.querySelector("#message");
+    const composer = document.querySelector("#composer");
+    const sendButton = document.querySelector("#send");
+    const connectionStatus = document.querySelector("#connection-status");
+
+    function appendMessage(role, text, state = "") {
+      const wrapper = document.createElement("article");
+      wrapper.className = `message ${role} ${state}`;
+      const avatar = document.createElement("span");
+      avatar.className = "avatar";
+      avatar.textContent = role === "user" ? "我" : "龙";
+      const bubble = document.createElement("p");
+      bubble.className = "bubble";
+      bubble.textContent = text;
+      wrapper.append(avatar, bubble);
+      chat.append(wrapper);
+      chat.scrollTop = chat.scrollHeight;
+      return { wrapper, bubble };
+    }
+
+    async function ensureVisitorToken() {
+      if (accessToken) return;
+      const response = await fetch("/api/v1/visitor-tokens", { method: "POST" });
+      if (!response.ok) throw new Error("访客令牌获取失败");
+      accessToken = (await response.json()).access_token;
+      connectionStatus.innerHTML = '<span class="role">访客</span>：测试连接已建立';
+    }
+
+    function delay(milliseconds) {
+      return new Promise((resolve) => window.setTimeout(resolve, milliseconds));
+    }
+
+    async function pollRun(statusUrl, messageView) {
+      for (let attempt = 0; attempt < 40; attempt += 1) {
+        const response = await fetch(statusUrl, { headers: { Authorization: `Bearer ${accessToken}` } });
+        if (!response.ok) throw new Error("客服运行状态读取失败");
+        const run = (await response.json()).data;
+        if (run.status === "succeeded") {
+          messageView.bubble.textContent = run.result?.content || "客服未返回可显示内容。";
+          messageView.wrapper.classList.remove("pending");
+          chat.scrollTop = chat.scrollHeight;
+          return;
+        }
+        if (run.status === "failed" || run.status === "cancelled") {
+          messageView.bubble.textContent = `本次客服运行未完成：${run.error_code || run.status}`;
+          messageView.wrapper.className = "message error";
+          return;
+        }
+        await delay(500);
+      }
+      messageView.bubble.textContent = "客服请求仍在处理中，请确认 Worker 已启动后重试。";
+      messageView.wrapper.className = "message error";
+    }
+
+    async function sendMessage(message) {
+      const normalized = message.trim();
+      if (!normalized) return;
+      appendMessage("user", normalized);
+      input.value = "";
+      sendButton.disabled = true;
+      const pending = appendMessage("agent", "奶龙基金智能助手正在处理您的问题…", "pending");
+      try {
+        await ensureVisitorToken();
+        const response = await fetch("/api/v1/agent-runs", {
+          method: "POST",
+          headers: { "Content-Type": "application/json", Authorization: `Bearer ${accessToken}` },
+          body: JSON.stringify({ agent_type: "customer_service", message: normalized, session_id: sessionId, idempotency_key: crypto.randomUUID().replaceAll("-", "") }),
+        });
+        if (!response.ok) throw new Error("客服请求创建失败");
+        await pollRun((await response.json()).data.status_url, pending);
+      } catch (error) {
+        pending.bubble.textContent = error instanceof Error ? error.message : "客服测试请求失败";
+        pending.wrapper.className = "message error";
+      } finally {
+        sendButton.disabled = false;
+        input.focus();
+      }
+    }
+
+    composer.addEventListener("submit", (event) => {
+      event.preventDefault();
+      void sendMessage(input.value);
+    });
+    document.querySelectorAll(".sample").forEach((button) => {
+      button.addEventListener("click", () => void sendMessage(button.dataset.message || ""));
+    });
+    appendMessage("agent", "您好呀，我是奶龙基金智能助手。这里可测试公开问题、账户入口、安全提示和人工服务路径。");
+    void ensureVisitorToken().catch(() => { connectionStatus.textContent = "访客：连接未建立，发送消息时将自动重试"; });
+  </script>
+</body>
+</html>

tests/unit/api/test_customer_service_test_page.py

@@ -0,0 +1,13 @@
+from fastapi.testclient import TestClient
+
+from app.main import create_app
+
+
+def test_customer_service_test_page_exposes_visitor_agent_flow() -> None:
+    with TestClient(create_app()) as client:
+        response = client.get("/customer-service-test/")
+
+    assert response.status_code == 200
+    assert "奶龙基金智能助手" in response.text
+    assert "/api/v1/visitor-tokens" in response.text
+    assert "/api/v1/agent-runs" in response.text

tests/unit/service/test_agent_persistence_handover.py

@@ -0,0 +1,121 @@
+from datetime import UTC, datetime
+from typing import Any
+
+import pytest
+
+from app.core.contracts import AgentResult, CoreResult, IntentResult
+from app.model.conversation import ConversationMessage
+from app.model.platform import AgentRun, DomainEventOutbox, HandoverTicket
+from app.service.agent_persistence_service import AgentPersistenceService
+
+
+class FakeTransaction:
+    async def __aenter__(self) -> None:
+        return None
+
+    async def __aexit__(
+        self, exc_type: object, exc_value: object, traceback: object
+    ) -> bool:
+        return False
+
+
+class FakeSession:
+    """仅收集持久化服务在同一事务内计划写入的 ORM 实体。"""
+
+    def __init__(self, run: AgentRun) -> None:
+        self.run = run
+        self.added: list[Any] = []
+        self.executed: list[Any] = []
+
+    def begin(self) -> FakeTransaction:
+        return FakeTransaction()
+
+    async def scalar(self, statement: object) -> AgentRun:
+        return self.run
+
+    def add(self, item: Any) -> None:
+        self.added.append(item)
+
+    async def flush(self) -> None:
+        for item in self.added:
+            if isinstance(item, ConversationMessage) and item.id is None:
+                item.id = 901
+
+    async def execute(self, statement: object) -> None:
+        self.executed.append(statement)
+
+
+def queued_run() -> AgentRun:
+    now = datetime.now(UTC).replace(tzinfo=None)
+    return AgentRun(
+        id=1,
+        run_id="run-transfer-1",
+        idempotency_id=2,
+        session_id="session-transfer-1",
+        user_id=7,
+        agent_type="customer_service",
+        trace_id="trace-transfer-1",
+        request_message_id=800,
+        status="queued",
+        created_at=now,
+        updated_at=now,
+    )
+
+
+def result(*, transfer_required: bool) -> AgentResult:
+    return AgentResult(
+        run_id="run-transfer-1",
+        result=CoreResult(
+            text="已为您转接人工客服。",
+            intent=IntentResult(intent="human_handover", confidence=1),
+            transfer_required=transfer_required,
+            transfer_reason="user_requested" if transfer_required else None,
+        ),
+    )
+
+
+def added_of(items: list[Any], model: type[Any]) -> list[Any]:
+    return [item for item in items if isinstance(item, model)]
+
+
+@pytest.mark.asyncio
+async def test_transfer_required_result_creates_pending_ticket_and_outbox_event() -> None:
+    """Agent 发起的转人工必须在完成运行的事务内留待处理工单和通知事件。"""
+    session = FakeSession(queued_run())
+
+    await AgentPersistenceService(session).complete_run(
+        "run-transfer-1", result(transfer_required=True), memory_extraction_requested=False
+    )
+
+    tickets = added_of(session.added, HandoverTicket)
+    events = added_of(session.added, DomainEventOutbox)
+    assert len(tickets) == 1
+    assert tickets[0].status == "pending"
+    assert tickets[0].session_id == "session-transfer-1"
+    assert tickets[0].customer_id == 7
+    assert tickets[0].source_agent == "customer_service"
+    assert tickets[0].source_message_id == 901
+    assert tickets[0].reason_code == "user_requested"
+    assert len(events) == 2  # agent.run_completed + conversation.transfer_requested
+    transfer_event = next(
+        event for event in events if event.event_type == "conversation.transfer_requested"
+    )
+    assert transfer_event.aggregate_type == "conversation"
+    assert transfer_event.aggregate_id == "session-transfer-1"
+    assert transfer_event.payload["ticket_no"] == tickets[0].ticket_no
+
+
+@pytest.mark.asyncio
+async def test_normal_result_does_not_create_handover_ticket_or_event() -> None:
+    """非转人工回答不得污染管理员待处理队列。"""
+    session = FakeSession(queued_run())
+
+    await AgentPersistenceService(session).complete_run(
+        "run-transfer-1", result(transfer_required=False), memory_extraction_requested=False
+    )
+
+    assert added_of(session.added, HandoverTicket) == []
+    assert all(
+        event.event_type != "conversation.transfer_requested"
+        for event in added_of(session.added, DomainEventOutbox)
+    )

tests/unit/service/test_customer_service_agent.py

@@ -0,0 +1,171 @@
+import pytest
+
+from app.core.contracts import AgentRequest, AgentRequestMetadata, RequestContext
+from app.core.errors import RecoverableAgentError
+from app.core.knowledge_contracts import KnowledgeHit, KnowledgeSearchResult
+from app.service.agent.base import BaseAgent
+from app.service.agent.bootstrap import get_agent_factory
+from app.service.agent.customer_service_agent import CustomerServiceAgent
+from app.service.agent.customer_service_routing import CustomerServiceIntentRouter
+
+
+def request(message: str, *, chitchat_streak: int = 0) -> AgentRequest:
+    return AgentRequest(
+        agent_type="customer_service",
+        message=message,
+        session_id="customer-service-session",
+        idempotency_key="customer-service-idempotency-key",
+        metadata=AgentRequestMetadata(chitchat_streak=chitchat_streak),
+    )
+
+
+def context(role: str) -> RequestContext:
+    return RequestContext(
+        user_id="1",
+        trace_id="customer-service-trace",
+        roles=(role,),
+        permissions=("agent:run", "knowledge:query"),
+        data_scope="public" if role == "visitor" else "self",
+    )
+
+
+@pytest.mark.asyncio
+async def test_visitor_account_question_only_returns_login_entry() -> None:
+    result = await CustomerServiceAgent().handle(
+        request("我的持仓收益是多少"), context("visitor")
+    )
+
+    assert result.text == "我无法查询账户数据，请先登录后前往“我的账户”查看相关状态。"
+    assert result.transfer_required is False
+
+
+@pytest.mark.asyncio
+async def test_authenticated_account_question_only_returns_account_entry() -> None:
+    result = await CustomerServiceAgent().handle(request("查一下我的订单"), context("customer"))
+
+    assert result.text == "我无法查询账户数据，请前往“我的账户”查看相关状态。"
+    assert result.transfer_required is False
+
+
+@pytest.mark.asyncio
+async def test_security_question_requires_human_transfer() -> None:
+    result = await CustomerServiceAgent().handle(
+        request("验证码已经发给别人了"), context("visitor")
+    )
+
+    assert result.transfer_required is True
+    assert result.transfer_reason == "security_notice"
+
+
+@pytest.mark.asyncio
+async def test_personalized_investment_advice_is_refused_and_transferred() -> None:
+    result = await CustomerServiceAgent().handle(
+        request("帮我推荐一只收益最高的基金"), context("customer")
+    )
+
+    assert result.transfer_required is True
+    assert result.transfer_reason == "compliance_refusal"
+
+
+def test_customer_service_agent_is_registered_for_visitor_role() -> None:
+    agent = get_agent_factory().create("customer_service", context("visitor"))
+
+    assert isinstance(agent, CustomerServiceAgent)
+
+
+def test_chitchat_streak_is_derived_from_continuous_prior_messages() -> None:
+    assert CustomerServiceIntentRouter.chitchat_streak(
+        ("你好", "讲个笑话", "你开心吗"), "在吗"
+    ) == 4
+    assert CustomerServiceIntentRouter.chitchat_streak(
+        ("你好", "基金怎么开户"), "在吗"
+    ) == 1
+
+
+@pytest.mark.asyncio
+async def test_policy_question_uses_only_policy_knowledge(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, object] = {}
+
+    async def fake_call_tool(
+        self: BaseAgent,
+        name: str,
+        arguments: dict[str, object],
+        *,
+        intent: str,
+        context: RequestContext,
+    ) -> KnowledgeSearchResult:
+        captured.update(name=name, arguments=arguments, intent=intent)
+        return KnowledgeSearchResult(
+            hits=(
+                KnowledgeHit(
+                    knowledge_id="101",
+                    collection="fin_policy_collection",
+                    snippet="赎回规则摘要",
+                    answer="这是已审核的赎回公开规则。",
+                ),
+            )
+        )
+
+    monkeypatch.setattr(BaseAgent, "call_tool", fake_call_tool)
+
+    result = await CustomerServiceAgent().handle(
+        request("基金赎回到账规则是什么"), context("customer")
+    )
+
+    assert result.text == "这是已审核的赎回公开规则。"
+    assert captured == {
+        "name": "query_knowledge",
+        "arguments": {
+            "query": "基金赎回到账规则是什么",
+            "intents": ("policy_explain",),
+            "top_k": 5,
+        },
+        "intent": "public_knowledge",
+    }
+
+
+@pytest.mark.asyncio
+async def test_knowledge_failure_requires_human_transfer(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def unavailable_tool(
+        self: BaseAgent,
+        name: str,
+        arguments: dict[str, object],
+        *,
+        intent: str,
+        context: RequestContext,
+    ) -> KnowledgeSearchResult:
+        raise RecoverableAgentError("知识检索不可用")
+
+    monkeypatch.setattr(BaseAgent, "call_tool", unavailable_tool)
+
+    result = await CustomerServiceAgent().handle(request("基金怎么开户"), context("visitor"))
+
+    assert result.transfer_required is True
+    assert result.transfer_reason == "knowledge_unavailable"
+
+
+@pytest.mark.asyncio
+async def test_fourth_chitchat_message_is_guided_once_without_knowledge_lookup(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def unexpected_tool(
+        self: BaseAgent,
+        name: str,
+        arguments: dict[str, object],
+        *,
+        intent: str,
+        context: RequestContext,
+    ) -> KnowledgeSearchResult:
+        raise AssertionError("闲聊不应调用公开知识工具")
+
+    monkeypatch.setattr(BaseAgent, "call_tool", unexpected_tool)
+
+    guided = await CustomerServiceAgent().handle(
+        request("你今天开心吗", chitchat_streak=4), context("visitor")
+    )
+    ordinary = await CustomerServiceAgent().handle(
+        request("你今天开心吗", chitchat_streak=5), context("visitor")
+    )
+
+    assert "基金业务" in guided.text
+    assert "基金业务" not in ordinary.text

tests/unit/service/test_customer_service_chitchat_metadata.py

@@ -0,0 +1,16 @@
+from app.core.contracts import AgentRequest, AgentRequestMetadata
+from app.service.agent_run_application_service import build_outbox_metadata
+
+
+def test_customer_service_outbox_metadata_overrides_supplied_chitchat_streak() -> None:
+    request = AgentRequest(
+        agent_type="customer_service",
+        message="在吗",
+        session_id="customer-service-session",
+        idempotency_key="customer-service-idempotency-key",
+        metadata=AgentRequestMetadata(chitchat_streak=5),
+    )
+
+    metadata = build_outbox_metadata(request, ("你好", "基金怎么开户"))
+
+    assert metadata["chitchat_streak"] == 1