group_fqcd_jr/docs/01-通用Agent平台开发设计.md

MVC+S 通用 Agent 底座开发设计

版本：v3.1
适用对象：通用底座负责人、业务 Agent 开发人员、编码 Agent
文档性质：实现规范；未特别标注“示例”的接口、字段和执行顺序均为强制约定 修订日期：2026-09-09

1. 文档目的

本文定义智能财富管理系统在 MVC+S 架构下的公共 Agent 运行底座，供客服、投顾、风控和运营 Agent 共同使用。建设策略是“厚底座、薄 Agent”：底座先完成鉴权、会话、幂等、长期记忆、Neo4j 关系推理、配置发布、多模型调度、工具、合规、审计和事件可靠投递，组员只继承 BaseAgent、填写声明式配置并实现领域 handle()。

本文覆盖：

• Python、FastAPI 和 SQLAlchemy 的 MVC+S 工程结构。
• BaseAgent 通用执行骨架。
• AgentFactory 注册与创建机制。
• Service 层内的模型、工具、记忆、合规、审计和事件接口。
• SSE 流式协议、异常处理、可观测性和测试门槛。

本文不覆盖以下内容：

• 各业务 Agent 的完整业务规则。
• 真实证券交易、充值、提现或场外基金交易实现。
• 前端页面设计。

1.1 强制演进原则

• 数据库以新版基线为权威源，允许新增表和新增字段。
• 禁止重命名任何已有表，禁止删除、重命名或复用任何已有字段。
• 禁止改变已有字段的数据类型、可空性和既有业务含义；发现历史设计问题时通过新增字段、新表、索引或应用兼容逻辑解决。
• 业务 Agent 不得要求修改公共底座；缺失的共性能力由底座负责人以向后兼容方式扩展。
• AgentDefinition 是业务 Agent 的权限上限和能力声明，不是基础设施实现入口。

2. 技术栈

领域	技术	用途
Web API	FastAPI	REST、SSE、鉴权依赖
数据访问	SQLAlchemy 2.x	ORM、事务、连接管理
数据校验	Pydantic 2.x	请求、响应和配置模型
主数据库	MySQL 8.0	权威业务数据、记忆和审计
缓存	Redis	短期会话、中期记忆热缓存、限流
向量检索	Milvus	知识库和长期语义记忆
关系存储	Neo4j	长期画像关系和实体关联
数据迁移	Alembic	版本化建表和升级回滚
测试	pytest	单元、集成、契约和安全测试

3. MVC+S 总体架构

3.1 分层职责

层	目录	职责	不允许承担的职责
Model	app/model	SQLAlchemy 实体、数据库字段和持久化映射	Agent 编排、HTTP 响应
View	app/view	统一 JSON 响应、SSE 事件和错误展示模型	业务判断、数据库访问
Controller	app/controller	路由、请求校验、调用 Service、返回 View	直接调用模型或编写 Agent 业务
Service	app/service	Agent 工厂、执行骨架、业务编排、记忆、工具、合规和事务	解析 HTTP 细节、拼装前端页面

Controller 只调用 Service；Service 通过 Repository 使用 Model；执行结果交给 View 转换为统一响应。Agent 是 Service 层的业务编排组件，不是独立于 MVC+S 的第五层。

3.2 调用关系

Controller / FastAPI Router
    |
    v
AgentService ---- RequestContextBuilder / RBAC
    |
    v
    AgentFactory ---- AgentRegistry ---- AgentDefinition
    |
    v
BaseAgent.execute()
    |-- 1. validate_input
    |-- 2. recall_memory
    |-- 3. classify_intent
    |-- 4. handle                <- 业务 Agent 实现
    |-- 5. generate_and_guard
    |-- 6. persist_and_audit
    `-- 7. publish_events
         |
         +-- ModelGateway
         +-- ToolExecutor
         +-- MemoryService
         +-- RelationshipService
         +-- ConfigCenter
         +-- ModelRouter
         +-- KnowledgeService
         +-- ComplianceService
         +-- AuditService
         `-- AgentPersistenceService / DomainEventOutbox
    |
    +--> Repository --> Model / MySQL
    `--> ViewBuilder --> JSON or SSE View

3.3 MVC+S 调用约束

Controller 保持薄层，只处理 HTTP；创建运行不在请求线程直接执行模型：

@router.post("/agent-runs")
async def create_run(
    request: AgentRunCreateRequest,
    context: RequestContext = Depends(build_request_context),
    service: AgentRunApplicationService = Depends(get_agent_run_service),
) -> JSONResponse:
    run = await service.accept(request, context)
    return AgentJsonView.response(run, status_code=202)

Service 负责应用编排：

class AgentExecutor:
    def __init__(self, factory: AgentFactory) -> None:
        self.factory = factory

    async def execute(
        self,
        agent_type: str,
        request: AgentRequest,
        context: RequestContext,
    ) -> AsyncIterator[RunProgressEvent]:
        agent = self.factory.create(agent_type, context)
        async for event in agent.execute(request, context):
            yield event

Model 只定义数据与映射，Repository 封装查询；View 只把运行投影或 RunProgressEvent 转为外部 JSON/SSE 格式。任何业务条件都不能写在 Controller 或 View 中。

4. 工程目录

app/
  main.py
  controller/
    deps.py
    agent_controller.py
    health_controller.py
  view/
    response.py
    sse.py
    errors.py
  model/
    base.py
    entities/
    repositories/
    dto/
  service/
    agent_service.py
    agent/
      base.py
      factory.py
      registry.py
      bootstrap.py
      authorizer.py
      contracts.py
      context.py
      definitions.py
      implementations/
        customer_service_agent.py
        advisor_agent.py
        risk_agent.py
        operations_agent.py
    model_gateway/
      base.py
      router.py
      policies.py
      health.py
      providers/
      qwen.py
    tool/
      contracts.py
      registry.py
      executor.py
    memory/
      service.py
      recall.py
      extraction.py
      promotion.py
      forgetting.py
      outbox.py
      graph_sync.py
    relationship/
      service.py
      schema.py
      query_templates.py
    config_center/
      service.py
      resolver.py
      publisher.py
      cache.py
    session/
      service.py
      idempotency.py
    knowledge/
      service.py
      milvus.py
      mysql_fallback.py
    compliance/
      service.py
      masking.py
      negative_words.py
      suitability.py
    audit/
      service.py
    event/
      publisher.py
      contracts.py
      outbox_worker.py
  core/
    config.py
    errors.py
    logging.py
    security.py
tests/
  unit/
  integration/
  contract/
  security/
alembic/
  versions/

依赖方向固定为：Controller -> Service -> Repository -> Model，Service -> View 仅返回中立结果，由 Controller 选择 JSON 或 SSE View。业务 Agent 位于 app/service/agent，不得直接创建数据库连接，也不得直接调用 Redis、Milvus 或 Neo4j 客户端。

5. 完整数据契约

5.1 命名与类型规则

• agent_type 使用小写蛇形字符串，例如 customer_service，不使用中心枚举；新增 Agent 不需要修改公共枚举文件。
• intent 使用小写蛇形字符串，例如 policy_explain。
• confidence 使用 Decimal，对外和入库范围均为 0.0000 至 1.0000。
• 所有集合使用不可变类型，避免 Agent 在执行过程中修改共享上下文。
• 所有跨层对象使用显式类型，不传递无结构的顶层 dict。

5.2 请求、上下文和配置

from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal
from enum import StrEnum
from datetime import datetime, timezone
from typing import Any, AsyncIterator, ClassVar, Literal, Mapping, Protocol, Sequence
from uuid import uuid4

from pydantic import BaseModel, ConfigDict, Field, field_validator


class AgentRequestMetadata(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    locale: Literal["zh-CN"] = "zh-CN"
    client_version: str | None = Field(default=None, max_length=32)
    ui_entry: str | None = Field(
        default=None,
        min_length=1,
        max_length=32,
        pattern=r"^[a-z][a-z0-9_]*$",
    )


class AgentRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    session_id: str = Field(min_length=1, max_length=64)
    message: str = Field(min_length=1, max_length=8000)
    idempotency_key: str = Field(min_length=8, max_length=64)
    end_session: bool = False
    metadata: AgentRequestMetadata = Field(default_factory=AgentRequestMetadata)

    @field_validator("message")
    @classmethod
    def reject_blank_message(cls, value: str) -> str:
        if not value.strip():
            raise ValueError("message must not be blank")
        return value


@dataclass(frozen=True)
class RequestContext:
    trace_id: str
    session_id: str
    user_id: int
    portal: str
    roles: frozenset[str]
    data_scope: str
    clarification_round: int = 0
    assigned_customer_ids: frozenset[int] = frozenset()


@dataclass(frozen=True)
class AgentDefinition:
    agent_type: str
    display_name: str
    allowed_roles: frozenset[str]
    allowed_portals: frozenset[str]
    allowed_tools: frozenset[str]
    supported_intents: frozenset[str]
    intent_descriptions: Mapping[str, str]
    default_classification_instruction: str
    default_model_policy: str
    default_temperature: float
    default_intent_threshold: Decimal
    max_clarification_rounds: int = 2
    compliance_policy: str = "default"
    memory_policy: str = "default"
    required_memory_views: frozenset[str] = frozenset()
    required_relationship_views: frozenset[str] = frozenset()


@dataclass(frozen=True)
class ResolvedAgentConfig:
    agent_type: str
    temperature: float
    intent_threshold: Decimal
    max_clarification_rounds: int
    allowed_tools_by_intent: Mapping[str, frozenset[str]]
    supported_intents: frozenset[str]
    intent_descriptions: Mapping[str, str]
    classification_instruction: str
    compliance_policy: str
    model_policy: str
    memory_policy: str
    required_memory_views: frozenset[str]
    required_relationship_views: frozenset[str]
    config_version: str

AgentRequest.session_id 必须与 RequestContext.session_id 一致。user_id、角色、入口、数据范围和 clarification_round 只来自服务端上下文，客户端不能提交或覆盖；end_session 只是结束当前会话的显式信号，不能触发其他业务写操作。

metadata 只接受 locale、client_version 和 ui_entry，额外键由 Pydantic 直接拒绝。权限、模型、工具白名单、客户数据范围和任意业务参数均不得通过 metadata 传入；这些值必须来自服务端认证上下文、Agent 定义或已审核的配置快照。

5.3 意图、记忆、工具和结果

class SseEventType(StrEnum):
    START = "start"
    TOOLS = "tools"
    DELTA = "delta"
    REPLACE = "replace"
    DONE = "done"
    ERROR = "error"


@dataclass(frozen=True)
class SseEvent:
    event: SseEventType
    trace_id: str
    data: Mapping[str, Any]


@dataclass(frozen=True)
class RunProgressEvent:
    """Service/Worker 内部进度事件，不绑定 HTTP 或 SSE。"""

    event: str
    trace_id: str
    data: Mapping[str, Any]


@dataclass(frozen=True)
class IntentResult:
    name: str
    confidence: Decimal
    entities: Mapping[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class SourceReference:
    collection: str
    doc_id: str
    title: str | None = None
    version: str | None = None


@dataclass(frozen=True)
class RecalledContext:
    recent_messages: tuple[str, ...]
    medium_term_memories: tuple[str, ...]
    long_term_memories: tuple[str, ...]
    authoritative_facts: Mapping[str, Any]
    relationship_facts: Mapping[str, tuple[Mapping[str, Any], ...]]
    token_count: int
    relationship_degraded: bool = False


@dataclass(frozen=True)
class ToolCallRecord:
    tool_name: str
    arguments: Mapping[str, Any]
    status: str
    elapsed_ms: int


@dataclass(frozen=True)
class ToolResult:
    safe_text: str
    safe_data: Mapping[str, Any]
    record: ToolCallRecord
    source_references: tuple[SourceReference, ...] = ()


@dataclass(frozen=True)
class CoreResult:
    direct_reply: str | None = None
    generation_material: tuple[str, ...] = ()
    source_references: tuple[SourceReference, ...] = ()
    tool_calls: tuple[ToolCallRecord, ...] = ()
    suggestions: tuple[str, ...] = ()
    transfer_required: bool = False
    transfer_reason: str | None = None
    memory_extraction_marker: bool = False


@dataclass(frozen=True)
class AgentResult:
    reply: str
    intent: IntentResult
    source_references: tuple[SourceReference, ...]
    tool_calls: tuple[ToolCallRecord, ...]
    suggestions: tuple[str, ...]
    transfer_required: bool
    transfer_reason: str | None
    memory_extraction_marker: bool = False
    degraded: bool = False
    degradation_reason: str | None = None

CoreResult.direct_reply 用于审核过的 FAQ 或固定模板；为空时由基础类根据 generation_material 调用模型生成。direct_reply 与 generation_material 必须恰好提供一个。来源引用必须来自工具或知识服务返回值，禁止模型自行构造。

5.4 依赖协议

class ModelGateway(Protocol):
    async def classify(
        self,
        *,
        agent_type: str,
        trace_id: str,
        message: str,
        context: RecalledContext,
        intents: frozenset[str],
        intent_descriptions: Mapping[str, str],
        classification_instruction: str,
        model_policy: str,
    ) -> IntentResult: ...

    async def generate(
        self,
        *,
        agent_type: str,
        trace_id: str,
        message: str,
        material: Sequence[str],
        context: RecalledContext,
        model_policy: str,
        temperature: float,
    ) -> str: ...


class MemoryService(Protocol):
    async def recall(
        self, context: RequestContext, message: str
    ) -> RecalledContext: ...


class ComplianceService(Protocol):
    async def validate_input(
        self, request: AgentRequest, context: RequestContext
    ) -> None: ...

    async def guard_output(
        self, text: str, policy: str, context: RequestContext
    ) -> str: ...


class ConversationService(Protocol):
    async def save_user_message(
        self, request: AgentRequest, context: RequestContext
    ) -> int: ...

class AuditService(Protocol):
    async def record_agent_run(
        self,
        *,
        context: RequestContext,
        agent_type: str,
        outcome: str,
        detail: Mapping[str, Any],
    ) -> None: ...

    async def record_unexpected_error(
        self,
        *,
        context: RequestContext,
        agent_type: str,
        error: Exception,
    ) -> str: ...


@dataclass(frozen=True)
class DomainEvent:
    event_id: str
    event_type: str
    aggregate_type: str
    aggregate_id: str
    trace_id: str
    payload: Mapping[str, Any]
    occurred_at: datetime


class AgentPersistenceService(Protocol):
    async def complete_run(
        self,
        *,
        result: AgentResult,
        context: RequestContext,
        agent_type: str,
        user_message_id: int,
        outcome: str,
        detail: Mapping[str, Any],
        memory_extraction_requested: bool,
    ) -> int:
        """在同一事务写助手消息、执行审计、幂等结果和领域事件Outbox。"""
        ...


class AgentConfigService(Protocol):
    async def resolve(
        self, definition: AgentDefinition
    ) -> ResolvedAgentConfig: ...


class FallbackService(Protocol):
    async def reply_for(self, error_code: str, policy: str) -> str: ...


class ToolExecutor(Protocol):
    async def execute(
        self,
        *,
        tool_name: str,
        arguments: Mapping[str, Any],
        allowed_tools: frozenset[str],
        context: RequestContext,
    ) -> ToolResult: ...


@dataclass(frozen=True)
class AgentDependencies:
    model_gateway: ModelGateway
    memory_service: MemoryService
    tool_executor: ToolExecutor
    compliance_service: ComplianceService
    conversation_service: ConversationService
    persistence_service: AgentPersistenceService
    audit_service: AuditService
    config_service: AgentConfigService
    fallback_service: FallbackService

Repository、Redis、Milvus 和 Neo4j 客户端隐藏在对应 Service 后面。业务 Agent 不依赖具体客户端类型。

5.5 错误分类

class AgentError(Exception):
    code = "AGENT_ERROR"
    public_message = "请求处理失败"


class AgentRequestError(AgentError):
    """参数、认证、权限或会话归属错误，不生成助手消息。"""


class AgentInputError(AgentRequestError):
    code = "AGENT_INPUT_INVALID"
    public_message = "请求参数不正确"


class AgentPermissionError(AgentRequestError):
    code = "AGENT_PERMISSION_DENIED"
    public_message = "无权访问该功能"


class RecoverableAgentError(AgentError):
    """模型、知识库或只读工具故障，可以使用安全模板降级。"""

    def __init__(self, code: str) -> None:
        super().__init__(code)
        self.code = code

底层异常必须在 Service 边界转换为上述类型。HTTP 状态码和错误 JSON 由 View 映射，业务 Agent 不返回 HTTP 响应。

6. BaseAgent 执行骨架

6.1 模板方法约束

execute() 是底座唯一公开执行入口，并使用 @final 标记。它只返回传输无关的运行结果和进度事件，不依赖 HTTP 或 SSE。业务子类只必须提供类级 definition 和实现 handle()；默认意图识别由基础类完成。确实需要自定义分类时，只能覆盖 _classify_intent() 保护钩子，不能覆盖整个执行流程。SseView 负责把进度事件映射为 01 §12 的 SSE 事件。

6.2 参考实现

from abc import ABC, abstractmethod
from dataclasses import asdict
from typing import AsyncIterator, ClassVar, final


class BaseAgent(ABC):
    definition: ClassVar[AgentDefinition]

    def __init__(self, dependencies: AgentDependencies) -> None:
        self._deps = dependencies
        self.model_gateway = dependencies.model_gateway
        self.memory_service = dependencies.memory_service
        self.tool_executor = dependencies.tool_executor
        self.compliance_service = dependencies.compliance_service
        self.conversation_service = dependencies.conversation_service
        self.persistence_service = dependencies.persistence_service
        self.audit_service = dependencies.audit_service
        self.config_service = dependencies.config_service
        self.fallback_service = dependencies.fallback_service

    @final
    async def execute(
        self,
        request: AgentRequest,
        context: RequestContext,
    ) -> AsyncIterator[RunProgressEvent]:
        yield RunProgressEvent(
            event="start",
            trace_id=context.trace_id,
            data={"session_id": context.session_id},
        )

        user_message_id: int | None = None
        config: ResolvedAgentConfig | None = None
        try:
            if request.session_id != context.session_id:
                raise AgentInputError(
                    "session_id does not match authenticated context"
                )
            config = await self.config_service.resolve(self.definition)
            await self.compliance_service.validate_input(request, context)
            user_message_id = await self.conversation_service.save_user_message(
                request, context
            )
            recalled = await self.memory_service.recall(context, request.message)
            intent = await self._classify_intent(
                request, context, recalled, config
            )
            core = await self._dispatch(request, context, recalled, intent, config)
            raw_reply = await self._build_reply(
                request, context, recalled, core, config
            )
            safe_reply = await self.compliance_service.guard_output(
                raw_reply, config.compliance_policy, context
            )
            result = AgentResult(
                reply=safe_reply,
                intent=intent,
                source_references=core.source_references,
                tool_calls=core.tool_calls,
                suggestions=core.suggestions,
                transfer_required=core.transfer_required,
                transfer_reason=core.transfer_reason,
                memory_extraction_marker=core.memory_extraction_marker,
            )
            memory_extraction_requested = should_extract_memory(
                request=request,
                recalled=recalled,
                result=result,
            )
            assistant_message_id = await self.persistence_service.complete_run(
                result=result,
                context=context,
                agent_type=self.definition.agent_type,
                user_message_id=user_message_id,
                outcome="success",
                detail={
                    "intent": intent.name,
                    "confidence": str(intent.confidence),
                    "config_version": config.config_version,
                },
                memory_extraction_requested=memory_extraction_requested,
            )

            if result.tool_calls:
                yield RunProgressEvent(
                    event="tools",
                    trace_id=context.trace_id,
                    data={"calls": serialize_tool_calls(result.tool_calls)},
                )
            for chunk in chunk_safe_text(result.reply):
                yield RunProgressEvent(
                    event="delta",
                    trace_id=context.trace_id,
                    data={"text": chunk},
                )
            yield self._done_event(result, context)

        except AgentRequestError as exc:
            await self.audit_service.record_agent_run(
                context=context,
                agent_type=self.definition.agent_type,
                outcome="rejected",
                detail={"error_code": exc.code},
            )
            yield RunProgressEvent(
                event="error",
                trace_id=context.trace_id,
                data={"error_code": exc.code, "message": exc.public_message},
            )

        except RecoverableAgentError as exc:
            try:
                degraded = await self._persist_degraded_result(
                    request=request,
                    context=context,
                    user_message_id=user_message_id,
                    error=exc,
                    config=config,
                )
            except Exception as fallback_exc:
                error_id = await self.audit_service.record_unexpected_error(
                    context=context,
                    agent_type=self.definition.agent_type,
                    error=fallback_exc,
                )
                yield RunProgressEvent(
                    event="error",
                    trace_id=context.trace_id,
                    data={
                        "error_code": "AGENT_INTERNAL_ERROR",
                        "message": "服务暂时不可用，请稍后重试",
                        "error_id": error_id,
                    },
                )
                return
            yield RunProgressEvent(
                event="replace",
                trace_id=context.trace_id,
                data={"text": degraded.reply},
            )
            yield self._done_event(degraded, context)

        except Exception as exc:
            error_id = await self.audit_service.record_unexpected_error(
                context=context,
                agent_type=self.definition.agent_type,
                error=exc,
            )
            yield RunProgressEvent(
                event="error",
                trace_id=context.trace_id,
                data={
                    "error_code": "AGENT_INTERNAL_ERROR",
                    "message": "服务暂时不可用，请稍后重试",
                    "error_id": error_id,
                },
            )

    async def _classify_intent(
        self,
        request: AgentRequest,
        context: RequestContext,
        recalled: RecalledContext,
        config: ResolvedAgentConfig,
    ) -> IntentResult:
        result = await self.model_gateway.classify(
            agent_type=self.definition.agent_type,
            trace_id=context.trace_id,
            message=request.message,
            context=recalled,
            intents=config.supported_intents,
            intent_descriptions=config.intent_descriptions,
            classification_instruction=config.classification_instruction,
            model_policy=config.model_policy,
        )
        if result.name not in config.supported_intents:
            raise RecoverableAgentError("UNKNOWN_INTENT")
        if not Decimal("0") <= result.confidence <= Decimal("1"):
            raise RecoverableAgentError("INVALID_CONFIDENCE")
        return result

    @abstractmethod
    async def handle(
        self,
        request: AgentRequest,
        context: RequestContext,
        recalled: RecalledContext,
        intent: IntentResult,
        config: ResolvedAgentConfig,
    ) -> CoreResult:
        ...

以下辅助方法属于 BaseAgent，是底座契约的一部分。业务子类不得覆盖：

    async def _dispatch(
        self,
        request: AgentRequest,
        context: RequestContext,
        recalled: RecalledContext,
        intent: IntentResult,
        config: ResolvedAgentConfig,
    ) -> CoreResult:
        if intent.confidence >= config.intent_threshold:
            return await self.handle(request, context, recalled, intent, config)

        if context.clarification_round < config.max_clarification_rounds:
            reply = await self.fallback_service.reply_for(
                "LOW_CONFIDENCE_CLARIFY",
                config.compliance_policy,
            )
            return CoreResult(direct_reply=reply)

        reply = await self.fallback_service.reply_for(
            "LOW_CONFIDENCE_TRANSFER",
            config.compliance_policy,
        )
        return CoreResult(
            direct_reply=reply,
            transfer_required=True,
            transfer_reason="intent_confidence_below_threshold",
        )

    async def _build_reply(
        self,
        request: AgentRequest,
        context: RequestContext,
        recalled: RecalledContext,
        core: CoreResult,
        config: ResolvedAgentConfig,
    ) -> str:
        has_direct = bool(core.direct_reply and core.direct_reply.strip())
        has_material = bool(core.generation_material)
        if has_direct == has_material:
            raise RecoverableAgentError("INVALID_CORE_RESULT")
        if has_direct:
            return core.direct_reply.strip()
        return await self.model_gateway.generate(
            agent_type=self.definition.agent_type,
            trace_id=context.trace_id,
            message=request.message,
            material=core.generation_material,
            context=recalled,
            model_policy=config.model_policy,
            temperature=config.temperature,
        )

    def _done_event(
        self,
        result: AgentResult,
        context: RequestContext,
    ) -> RunProgressEvent:
        return RunProgressEvent(
            event="done",
            trace_id=context.trace_id,
            data={
                "intent": result.intent.name,
                "confidence": str(result.intent.confidence),
                "source_references": [
                    asdict(reference) for reference in result.source_references
                ],
                "suggestions": list(result.suggestions),
                "transfer_required": result.transfer_required,
                "degraded": result.degraded,
            },
        )

    async def _persist_degraded_result(
        self,
        *,
        request: AgentRequest,
        context: RequestContext,
        user_message_id: int | None,
        error: RecoverableAgentError,
        config: ResolvedAgentConfig | None,
    ) -> AgentResult:
        if user_message_id is None:
            user_message_id = await self.conversation_service.save_user_message(
                request, context
            )
        reply = await self.fallback_service.reply_for(
            error.code,
            config.compliance_policy if config else self.definition.compliance_policy,
        )
        result = AgentResult(
            reply=reply,
            intent=IntentResult(name="unknown", confidence=Decimal("0.0000")),
            source_references=(),
            tool_calls=(),
            suggestions=(),
            transfer_required=False,
            transfer_reason=None,
            degraded=True,
            degradation_reason=error.code,
        )
        await self.persistence_service.complete_run(
            result=result,
            context=context,
            agent_type=self.definition.agent_type,
            user_message_id=user_message_id,
            outcome="degraded",
            detail={
                "error_code": error.code,
            },
            memory_extraction_requested=False,
        )
        return result

FallbackService 只能返回已经审核和版本化的模板；降级分支优先使用本次解析出的 config.compliance_policy。只有配置解析本身失败、尚未形成配置快照时，才使用 AgentDefinition 的安全默认策略。降级分支不得把异常文本、模型原文或工具原始响应返回给用户。模板查找失败属于不可恢复异常，进入统一 AGENT_INTERNAL_ERROR 分支。

下列纯函数放在公共底座模块，便于独立测试：

MEMORY_CONTEXT_TOKEN_THRESHOLD = 6000


def should_extract_memory(
    *,
    request: AgentRequest,
    recalled: RecalledContext,
    result: AgentResult,
) -> bool:
    return any(
        (
            request.end_session,
            recalled.token_count >= MEMORY_CONTEXT_TOKEN_THRESHOLD,
            result.transfer_required,
            result.memory_extraction_marker,
        )
    )


def build_domain_events(
    result: AgentResult,
    context: RequestContext,
    assistant_message_id: int,
    memory_extraction_requested: bool,
) -> tuple[DomainEvent, ...]:
    occurred_at = datetime.now(timezone.utc)
    events = [
        DomainEvent(
            event_id=str(uuid4()),
            event_type="conversation.completed",
            aggregate_type="conversation",
            aggregate_id=context.session_id,
            trace_id=context.trace_id,
            payload={
                "assistant_message_id": assistant_message_id,
                "agent_intent": result.intent.name,
                "degraded": result.degraded,
            },
            occurred_at=occurred_at,
        )
    ]
    if result.transfer_required:
        events.append(
            DomainEvent(
                event_id=str(uuid4()),
                event_type="conversation.transfer_requested",
                aggregate_type="conversation",
                aggregate_id=context.session_id,
                trace_id=context.trace_id,
                payload={
                    "assistant_message_id": assistant_message_id,
                    "reason": result.transfer_reason,
                },
                occurred_at=occurred_at,
            )
        )
    if memory_extraction_requested:
        events.append(
            DomainEvent(
                event_id=str(uuid4()),
                event_type="memory.extraction_requested",
                aggregate_type="conversation",
                aggregate_id=context.session_id,
                trace_id=context.trace_id,
                payload={"assistant_message_id": assistant_message_id},
                occurred_at=occurred_at,
            )
        )
    return tuple(events)


def chunk_safe_text(text: str, chunk_size: int = 256) -> tuple[str, ...]:
    if chunk_size < 1:
        raise ValueError("chunk_size must be positive")
    return tuple(
        text[offset : offset + chunk_size]
        for offset in range(0, len(text), chunk_size)
    )

should_extract_memory() 不是“每轮都提取”：仅在客户端明确结束会话、召回上下文达到阈值、发生转人工，或业务 handle() 返回明确沉淀标记时触发。memory_extraction_marker 必须表示已确认的业务节点，例如客户确认投资目标；不得因为普通问答或模型自行推断而设置。

AgentPersistenceService.complete_run() 在获得助手消息 ID 后、事务提交前调用 build_domain_events()。当 memory_extraction_requested=True 时，memory.extraction_requested 与助手消息、interaction_audit、request_idempotency 完成状态及其他 domain_event_outbox 事件在同一个 MySQL 短事务中写入；不得在提交后从请求线程直接调用记忆提取。该方法成功返回后才允许发送最终 SSE，事务外 Worker 再消费 Outbox 并执行记忆提取或消息中间件投递。

6.3 七步执行顺序

1. 输入与权限校验：校验消息长度、会话归属、角色、数据范围和敏感输入。
2. 记忆召回：读取当前会话、有效中期记忆和必要的长期记忆，最后以 MySQL 权威事实校验。
3. 意图路由：默认由 BaseAgent 通用分类器完成；低置信按配置进入澄清或转人工，特殊业务经审核后才覆盖分类钩子。
4. 核心逻辑：调用业务 Agent 的 handle() 钩子，例如知识检索或预警证据查询。
5. 结果生成与合规校验：生成回复、脱敏、检查禁止表达、注入必要免责声明。
6. 数据沉淀：保存用户消息、最终回答、意图、置信度、引用、工具调用和审计记录。
7. 事件广播：发布转人工、敏感意图、会话结束和记忆提取事件。

第 6、7 步由基础类强制执行。即使模型失败或采用降级回复，也必须保存最终实际返回给用户的内容。

6.4 执行约束

• 一个请求只生成一个 trace_id，贯穿 API、模型、工具、数据库和事件日志。
• 数据库写入采用短事务；模型和外部服务调用不得放在数据库事务内。
• 工具调用结果默认不直接返回，必须经过字段级权限过滤和脱敏。
• 客户端断开 SSE 后，服务端仍应完成必要的归档和审计。
• 同一个 trace_id 的重复请求返回已有结果或明确的处理中状态，不能重复创建工单。
• 幂等范围固定为 user_id + agent_type + idempotency_key；同键不同 request_hash 返回 409，相同已完成请求返回原助手消息。
• RequestContextBuilder 从 svc_conversation_session 读取澄清轮次；返回澄清问题时使用旧值条件原子递增，Redis 只做缓存。
• conversation_message 保存的是合规处理后的最终回复；被拦截的原始生成文本只允许保存不可逆摘要和命中规则编号。
• 在输出合规检查完成前不发送 delta，避免已经泄漏的内容只能依赖 replace 撤回。
• AgentRequestError 表示认证、授权或参数错误，不生成助手回复；可恢复依赖故障使用安全模板并完整留痕；未知异常只返回公共错误信息。

7. AgentFactory 与统一 Agent 格式

7.1 注册机制

import re


class AgentRegistry:
    def __init__(self) -> None:
        self._classes: dict[str, type[BaseAgent]] = {}

    def register(
        self,
        agent_type: str,
        agent_class: type[BaseAgent],
    ) -> None:
        if not re.fullmatch(r"[a-z][a-z0-9_]{1,31}", agent_type):
            raise InvalidAgentName(agent_type)
        if not issubclass(agent_class, BaseAgent):
            raise InvalidAgentClass(agent_class)
        if agent_class.definition.agent_type != agent_type:
            raise AgentDefinitionMismatch(agent_type)
        if agent_type in self._classes:
            raise DuplicateAgentRegistration(agent_type)
        self._classes[agent_type] = agent_class

    def get(self, agent_type: str) -> type[BaseAgent]:
        try:
            return self._classes[agent_type]
        except KeyError as exc:
            raise UnknownAgentName(agent_type) from exc

    def items(self) -> tuple[tuple[str, type[BaseAgent]], ...]:
        return tuple(self._classes.items())

注册在应用启动阶段显式完成，避免自动扫描目录造成导入顺序不确定：

def register_agents(registry: AgentRegistry) -> None:
    registry.register(CustomerServiceAgent.definition.agent_type, CustomerServiceAgent)
    registry.register(AdvisorAgent.definition.agent_type, AdvisorAgent)
    registry.register(RiskAgent.definition.agent_type, RiskAgent)
    registry.register(OperationsAgent.definition.agent_type, OperationsAgent)

工厂统一完成实例化和访问检查：

class AgentFactory:
    def __init__(
        self,
        registry: AgentRegistry,
        dependencies: AgentDependencies,
        authorizer: AgentAuthorizer,
    ) -> None:
        self.registry = registry
        self.dependencies = dependencies
        self.authorizer = authorizer

    def create(
        self,
        agent_type: str,
        context: RequestContext,
    ) -> BaseAgent:
        agent_class = self.registry.get(agent_type)
        self.authorizer.ensure_allowed(agent_class.definition, context)
        agent = agent_class(self.dependencies)
        return agent

7.2 工厂职责

AgentFactory.create(agent_type, context) 从注册表取得类，注入统一的 AgentDependencies 后创建实例，并执行以下检查：

1. agent_type 已注册。
2. 当前 portal 允许调用该 Agent。
3. 当前角色拥有 Agent 访问权限。
4. Agent 类继承 BaseAgent，且提供合法的 AgentDefinition。
5. 工厂注入的是请求级或无状态依赖，不复用带用户状态的 Agent 实例。
6. Agent 的模型、工具白名单和合规策略来自统一定义或版本化配置。

建议权限映射：

Agent	允许角色	数据范围
客服	customer、operator、admin	本人或职责范围内客户
投顾	advisor、admin	own_customers
风控	risk_operator、admin	按风控权限配置
运营	operator、admin	按运营权限配置

未注册类型和越权调用必须失败，不能自动回退到权限更高的 Agent。

7.3 配置权威性和覆盖规则

配置按以下顺序解析，不能由业务 Agent 自行改变：

1. 平台安全硬约束：跨客户隔离、只读工具限制、敏感字段和适当性规则，优先级最高，数据库配置不能关闭。
2. AgentDefinition 权限上限：允许角色、入口、支持意图、记忆视图、关系视图和工具是代码级上限；数据库只能缩小范围，不能扩大。
3. 已审核且生效的数据库配置：配置中心按一个 release_id 解析 agent_intent_config、平台配置、模型路由和 Prompt 版本；只有审核通过并完整发布的版本可生效。
4. AgentDefinition 默认值：数据库没有有效版本时使用，保证冷启动可运行。
5. 环境变量：只配置连接地址、凭证、超时上限等基础设施参数，不保存业务意图和合规话术。

客户端请求和 metadata 永远不能覆盖模型、阈值、工具白名单或合规策略。AgentConfigService.resolve() 每次请求只解析一次并生成不可变 ResolvedAgentConfig；采用的 config_version 必须写入审计。

数据库工具集合按当前意图计算交集，禁止先求 Agent 级并集再执行：

resolved_tools_by_intent[intent_code] = (
    definition.allowed_tools
    & database_config.allowed_tools_by_intent[intent_code]
    & role_permission.allowed_tools
)

任何一层未授权的工具都不能调用。数据库配置不能增加 AgentDefinition.supported_intents 之外的新意图；新增意图必须先提交代码声明和契约测试。

7.4 配置中心发布与回滚

配置中心管理 Agent 运行参数、意图路由、记忆策略、关系视图、Prompt、回复模板和模型路由。发布流程固定为：

创建草稿 -> Schema 校验 -> 安全上限校验 -> 双人审核
-> 生成不可变 release_id -> 原子激活 -> Redis 缓存失效
-> 新请求读取新版本 -> 指标观察 -> 必要时回滚上一版本

• 一个请求只使用一个配置快照，不允许执行中途切换版本。
• 发布失败时旧版本继续有效，不能出现半发布状态。
• 回滚是重新激活历史不可变版本，不直接修改历史记录。
• 配置缓存键为 agent-config:{agent_type}:{release_id}；缓存未命中回源 MySQL。
• 每次执行把 release_id、意图配置版本、Prompt 版本和模型路由版本写入审计。
• 配置中心不可用时使用最近一次已验证快照；本地没有快照时使用 AgentDefinition 默认值，但安全硬约束永不降级。

8. 工具系统

每个工具必须声明：名称、版本、输入模型、输出模型、超时、是否只读、允许的 Agent、所需权限和审计级别。

class AgentTool(Protocol):
    name: str
    read_only: bool
    timeout_seconds: float

    async def invoke(
        self,
        arguments: BaseModel,
        context: RequestContext,
    ) -> ToolResult:
        ...

ToolExecutor 必须在调用前后执行参数校验、白名单校验、权限检查、超时控制、结果脱敏和审计。NL2SQL 只能生成只读查询，并在服务端强制注入客户范围，不能依赖模型自行添加 customer_id 条件。

9. 模型网关

模型网关向业务层隐藏供应商差异，统一提供意图分类、文本生成、结构化生成、摘要和 Embedding 接口。业务 Agent 只声明 model_policy 和任务类型，不提交供应商、接口地址或密钥。

9.1 调度输入与输出

模型路由器的输入至少包含：agent_type、task_type、model_policy、数据敏感级别、最大延迟、最大成本、是否要求结构化输出和请求 trace_id。输出 ModelExecution 必须记录：实际端点、模型名、路由规则版本、Prompt 版本、尝试次数、Token、耗时、结束原因和是否降级。

任务类型固定使用：intent_classification、answer_generation、structured_extraction、conversation_summary、memory_extraction、relationship_explanation 和 embedding。新增任务类型由底座负责人注册，业务 Agent 不使用任意字符串绕过策略。

9.2 路由与故障转移

过滤不支持任务或敏感级别的端点
-> 过滤熔断、超配额和健康检查失败端点
-> 按发布中的路由规则匹配 agent_type + task_type
-> 在延迟、质量和成本预算内选择主模型
-> 幂等任务按规则有限重试
-> 切换已配置备用模型
-> 全部失败时返回审核模板或明确阻断

• 所有结构化结果使用 Pydantic 校验。
• 单次模型调用默认超时 15 秒；分类最多重试 1 次，生成不在同一端点盲目重试。
• 熔断窗口、失败阈值、半开探测和备用链由配置中心发布。
• 备用链以 model_routing_fallback 为权威来源并按 fallback_order 升序执行；model_routing_rule.fallback_endpoint_ids 仅是兼容快照，路由器不得据此绕过端点外键和顺序约束。
• 同一请求最多尝试 3 个模型端点，禁止无限重试。
• 全部失败时返回经过审核的模板，不暴露连接信息。
• Prompt、模型名和参数必须版本化并写入审计详情。
• 不在日志中记录未脱敏的完整客户资料。

9.3 模型端点安全

• 数据库只保存密钥引用 secret_ref，实际凭证由环境变量或密钥服务提供。
• 供应商不允许处理的敏感级别必须在路由前阻断。
• 健康检查只验证连通性和最小推理，不发送真实客户数据。
• 成本预算是保护措施，不能为了低成本绕过合规或使用未批准模型。

10. 长期记忆与 Neo4j 关系推理

10.1 记忆状态机

candidate -> verified -> promoted -> active
    |            |          |          |
    +-> rejected +-> conflict          +-> expired/deleted

1. 会话结束、上下文超限、转人工或明确业务节点触发异步提取。
2. 模型只产生候选；MemoryService 对类型、来源、证据、置信度和敏感级别做确定性校验。
3. 每个候选至少绑定一条 memory_evidence，无来源内容不得晋升。
4. 与现有记忆冲突时写 memory_conflict，不覆盖旧值。
5. 达到策略要求后生成新的 profile_snapshots，并与 memory_sync_outbox 在同一 MySQL 事务提交。
6. Milvus 和 Neo4j 消费者按画像版本幂等同步，低版本事件不能覆盖高版本。
7. 删除先在 MySQL 写失效和墓碑，再异步清理 Redis、Milvus 与 Neo4j。

MySQL 始终保存权威状态、证据和版本；Redis 是缓存，Milvus 是语义索引，Neo4j 是关系投影，三者都不能反向覆盖 MySQL 正式风险测评、交易、持仓和产品事实。

10.2 统一召回契约

MemoryService.recall() 根据 AgentDefinition.required_memory_views 和 required_relationship_views 统一编排 Redis、MySQL、Milvus 与 Neo4j。业务 Agent 只读取 RecalledContext，不得自行拼接多存储结果。

召回顺序固定为：租户和客户范围过滤、有效期过滤、语义或关系召回、去重、冲突过滤、MySQL 权威事实校验、Token 预算裁剪。每条长期记忆和关系事实必须携带来源、版本、有效期和置信度，缺少溯源的数据不进入模型上下文。

10.3 Neo4j 图模型

允许的基础节点为 Customer、Product、RiskProfile、Preference、Goal、Holding、Transaction、RiskAlert 和 KnowledgeTopic。允许的关系由 relationship/schema.py 白名单管理，例如 HAS_PROFILE、PREFERS、HAS_GOAL、HOLDS、TRADED、TRIGGERED 和 RELATED_TO。

• Neo4j 节点只保存业务主键、画像版本和检索所需的最小投影，不保存完整证件号、银行卡号或原始对话。
• 查询只能使用审核过的参数化 Cypher 模板，默认深度不超过 2 跳、结果不超过 100 条、超时不超过 2 秒。
• 禁止把模型生成的任意 Cypher 直接交给数据库执行。
• 所有查询先注入 user_id、数据范围和画像版本，返回后再做字段级脱敏。
• 推理结果属于辅助上下文；涉及余额、持仓、交易和风险等级时必须回查 MySQL。

10.4 关系服务接口

底座提供 RelationshipService.resolve_views(context, view_names)，视图名由代码注册，例如 customer_goal_context、portfolio_exposure、related_risk_events。组员只在 AgentDefinition.required_relationship_views 中声明视图，不编写 Cypher。Neo4j 不可用时返回空关系视图并标记 relationship_degraded=True，不能编造关系。

11. 合规服务

公共合规服务包含：

• 身份证、手机号、银行卡号、邮箱和真实姓名脱敏。
• 禁止表达的精确、变体和正则匹配。
• 客户与产品 C1-C5/R1-R5 适当性校验。
• 对客内容免责声明和 AI 生成标识。
• 输出中的跨客户数据和未授权字段检查。

合规失败分为两类：可重生成错误最多重试一次；越权、适当性、敏感数据泄露等硬错误直接拦截并写高优先级审计。

12. SSE 协议

事件	数据	含义
start	trace_id、session_id	请求已接受
tools	已脱敏工具调用摘要	工具阶段完成
delta	文本片段	增量回复
replace	完整安全回复	原输出不合规或模型异常时替换
done	intent、confidence、sources、suggestions	正常结束
error	error_code、message、trace_id	无法降级的请求错误

MVP 默认采用“全量生成、全量合规、持久化后分块发送”的合规优先模式，因此 delta 是安全文本分块，不承诺模型 Token 级真流式。若后续采用边生成边检查，必须另行设计完整句子缓冲和句级合规，不能把后来需要撤回的敏感片段提前发给客户端。

13. 异常和降级

故障	处理
Redis 不可用	会话降级至受限的进程缓存；权威事实继续查 MySQL
Milvus 超时	使用 fin_knowledge_meta.content_text 和标签做 MySQL 降级检索
Neo4j 不可用	跳过关系扩展，不影响 MySQL 权威事实和中期记忆
模型失败	有限重试、备用模型、审核模板三级降级
工具超时	记录超时并生成不包含未确认数据的回复
审计写入失败	对受监管业务停止返回成功结果；进入重试队列并告警
事件发布失败	通过 Outbox 异步重试，不能丢失关键事件

14. 可观测性

至少记录以下指标：

• 端到端 P50、P95 和 P99 延迟。
• 首个安全文本片段延迟，包括生成、合规和持久化耗时。
• 各意图数量、准确率和低置信率。
• 转人工率、转人工原因和处理时长。
• 各模型、工具和存储的成功率及耗时。
• 知识检索命中率、MySQL 降级次数和过期知识拦截次数。
• 合规拦截、越权拦截和敏感字段脱敏次数。
• 记忆召回、晋升、冲突和同步积压。

日志使用结构化 JSON，禁止记录密码、令牌、完整证件号、完整银行卡号和未经脱敏的模型上下文。

15. 测试要求

15.1 单元测试

• 工厂注册、重复注册、未知类型和角色拒绝。
• BaseAgent 七步顺序以及异常时仍执行归档。
• 工具白名单、参数、超时和数据范围校验。
• 置信度边界、合规重生成和安全替换。
• 记忆权威性、冲突和 Token 裁剪。
• 会话澄清轮次条件更新和 Redis 丢失恢复。
• 配置权限上限、发布校验、快照固定和回滚。
• 模型路由匹配、敏感级别过滤、重试上限和备用链。
• Neo4j 视图白名单、深度、结果数和超时限制。

15.2 集成测试

• FastAPI、MySQL、Redis、Milvus、Neo4j 的组合测试。
• 对话与审计记录内容一致。
• Outbox 部分失败和版本幂等。
• SSE 客户端中断后的数据沉淀。
• MySQL 降级检索只返回已发布且有效的知识。
• 助手消息、审计、幂等完成状态和领域事件在一个事务中提交。
• 模型、Prompt 和配置版本可通过 trace_id 完整还原。
• MySQL画像版本可幂等同步至 Milvus 和 Neo4j，并拒绝低版本覆盖。

15.3 安全测试

• 跨客户查询、伪造角色和绕过客户归属。
• Prompt 注入和工具参数注入。
• SQL 注入、NL2SQL 写语句和无条件全表查询。
• 敏感信息在响应、日志、审计和向量库中的泄漏。
• 模型端点越权、任意 Cypher、配置扩大权限和伪造关系视图。

核心公共逻辑单元测试覆盖率不得低于 80%，鉴权、适当性、跨客户隔离和禁止表达用例必须 100% 通过。

16. 业务 Agent 接入规范

16.1 组员需要完成的内容

新增业务 Agent 时，组员只需要提交以下内容：

1. 在 app/service/agent/implementations 新增一个 Agent 子类文件。
2. 提供 AgentDefinition，声明角色、入口、模型、阈值和工具白名单。
3. 配置 supported_intents，默认复用底座意图分类器。
4. 实现 handle()，只处理本领域核心逻辑。
5. 仅在通用分类器无法满足业务时，经底座负责人确认后覆盖 _classify_intent()。
6. 在统一启动注册函数中增加一行注册代码，并提交单元测试和契约测试数据。

不得复制 BaseAgent 的记忆、模型、审计、SSE 或异常处理代码。确有公共能力缺失时，应扩展公共接口并为所有 Agent 保持兼容。

16.2 统一子类模板

文件：app/service/agent/implementations/risk_agent.py

from decimal import Decimal
from typing import ClassVar

from app.service.agent.base import BaseAgent
from app.service.agent.contracts import (
    AgentDefinition,
    AgentRequest,
    CoreResult,
    IntentResult,
    RecalledContext,
    RequestContext,
    ResolvedAgentConfig,
)


class RiskAgent(BaseAgent):
    definition: ClassVar[AgentDefinition] = AgentDefinition(
        agent_type="risk",
        display_name="风控 Agent",
        allowed_roles=frozenset({"risk_operator", "admin"}),
        allowed_portals=frozenset({"risk"}),
        allowed_tools=frozenset({"query_alert", "query_risk_evidence"}),
        supported_intents=frozenset(
            {"general_risk_question", "alert_analysis", "handover_summary"}
        ),
        intent_descriptions={
            "general_risk_question": "不绑定具体预警的通用风控问题",
            "alert_analysis": "对指定预警及其证据进行辅助研判",
            "handover_summary": "生成供人工处理的上下文摘要",
        },
        default_classification_instruction=(
            "只根据用户问题和已授权上下文选择一个意图，并返回 JSON。"
        ),
        default_model_policy="balanced",
        default_temperature=0.3,
        default_intent_threshold=Decimal("0.6000"),
    )

    async def handle(
        self,
        request: AgentRequest,
        context: RequestContext,
        recalled: RecalledContext,
        intent: IntentResult,
        config: ResolvedAgentConfig,
    ) -> CoreResult:
        if intent.name == "general_risk_question":
            return CoreResult(
                generation_material=("请依据通用风控制度进行客观解释。",)
            )

        if intent.name == "handover_summary":
            return CoreResult(
                generation_material=(
                    "根据已授权的会话上下文生成客观交接摘要，不作处置结论。",
                ),
                transfer_required=True,
                transfer_reason="manual_risk_review",
            )

        alert_id_value = intent.entities.get("alert_id")
        if alert_id_value is None:
            reply = await self.fallback_service.reply_for(
                "MISSING_ALERT_ID",
                config.compliance_policy,
            )
            return CoreResult(direct_reply=reply)

        alert_id = int(alert_id_value)
        tool = await self.tool_executor.execute(
            tool_name="query_risk_evidence",
            arguments={"alert_id": alert_id},
            allowed_tools=config.allowed_tools_by_intent.get(
                intent.name,
                frozenset(),
            ),
            context=context,
        )
        return CoreResult(
            generation_material=(tool.safe_text,),
            source_references=tool.source_references,
            tool_calls=(tool.record,),
        )

组员不得重写 execute()。该方法是模板方法，负责固定执行顺序和公共治理。公共测试需要扫描所有 Agent 子类，发现覆盖 execute() 时直接失败。业务 Agent 不得直接构造 SseEvent。

16.3 统一契约测试

每个 Agent 子类自动参加同一组测试：

• definition.agent_type 与注册键一致。
• 角色、入口和工具集合不为空且引用合法配置。
• 基础分类器返回的置信度在 0 至 1。
• handle() 不能调用白名单之外的工具。
• 正常、低置信、模型失败和工具失败均产生最终消息和审计记录。
• 子类没有覆盖 execute()、记忆沉淀或事件广播方法。

最低契约测试示例：

import inspect

import pytest

from app.service.agent.base import BaseAgent
from app.service.agent.bootstrap import build_agent_registry
from app.core.errors import AgentPermissionError


def test_all_registered_agents_follow_base_contract() -> None:
    registry = build_agent_registry()
    protected_methods = {
        "execute",
        "_dispatch",
        "_build_reply",
        "_done_event",
        "_persist_degraded_result",
    }

    for key, agent_class in registry.items():
        assert issubclass(agent_class, BaseAgent)
        assert agent_class.definition.agent_type == key
        assert agent_class.definition.allowed_roles
        assert agent_class.definition.allowed_portals
        assert agent_class.definition.supported_intents
        assert protected_methods.isdisjoint(agent_class.__dict__)
        assert inspect.iscoroutinefunction(agent_class.handle)


def test_factory_rejects_role_without_access(
    agent_factory,
    customer_context,
) -> None:
    with pytest.raises(AgentPermissionError):
        agent_factory.create("risk", customer_context)

16.4 新增 Agent 文件清单

组员新增一个业务 Agent 时，变更范围应当限定为：

Create  app/service/agent/implementations/<agent_type>_agent.py
Modify  app/service/agent/bootstrap.py              # 增加一行注册
Create  tests/unit/service/agent/test_<agent_type>_agent.py
Modify  tests/contract/test_registered_agents.py    # 通常只增加参数数据
Optional Alembic/seed                               # 需要动态意图配置时

Controller、BaseAgent、AgentFactory、SSE View 和公共记忆服务通常不应因新增业务 Agent 而修改。

16.5 编码 Agent 交接模板

底座负责人把下面模板作为任务正文交给编码 Agent。尖括号内容必须在派发前替换，不能把未确定项留给编码 Agent 猜测。

任务：新增 <agent_type> 业务 Agent

架构约束：
- 遵守 MVC+S；本任务只实现 Service 层业务 Agent。
- 继承 BaseAgent，使用类级 AgentDefinition，只实现 handle()。
- 默认使用 BaseAgent._classify_intent()；未经底座负责人批准不得覆盖。
- 所有工具必须通过 self.tool_executor.execute() 调用。

允许创建或修改：
- app/service/agent/implementations/<agent_type>_agent.py
- app/service/agent/bootstrap.py（只增加注册）
- tests/unit/service/agent/test_<agent_type>_agent.py
- tests/contract/test_registered_agents.py（只增加该 Agent 参数）
- <经批准的配置 seed 或迁移文件；没有则写“无”>

禁止修改：
- app/service/agent/base.py
- app/service/agent/factory.py
- app/service/agent/contracts.py
- app/controller/**
- app/view/**
- app/service/memory/**
- 现有 Agent 的业务逻辑
- 数据库表结构（除非允许文件中明确列出迁移）

业务输入：
- agent_type: <小写蛇形名称>
- allowed_roles: <角色集合>
- allowed_portals: <入口集合>
- supported_intents: <意图集合及逐项定义>
- allowed_tools: <只读/写入工具清单及参数契约>
- compliance_policy: <策略名>
- model_policy: <模型调度策略名>
- required_memory_views: <记忆视图集合>
- required_relationship_views: <关系视图集合或“无”>
- intent_threshold: <0 至 1 的 Decimal>
- 每个意图的预期 CoreResult: <direct_reply 或 generation_material>
- 转人工条件: <明确条件>
- memory_extraction_marker 条件: <明确业务节点或“无”>

必须测试：
- 每个 supported_intent 至少一个正常用例。
- 阈值等于、低于 intent_threshold 的边界用例。
- 工具成功、超时、越权和返回空数据。
- 非允许角色和入口被工厂拒绝。
- 生成内容经过合规服务，来源引用来自工具结果。
- 子类未覆盖 execute() 及底座辅助方法。
- 降级分支保存最终消息并写审计。

验收命令：
python -m pytest tests/unit/service/agent/test_<agent_type>_agent.py -q
python -m pytest tests/contract/test_registered_agents.py -q
python -m pytest tests/security -q
python -m ruff check app tests
python -m mypy app

完成回报：
- 列出实际修改文件。
- 列出每个意图对应的处理分支和工具。
- 粘贴验收命令的退出码与测试数量。
- 说明未完成项；没有则明确写“无”。

编码 Agent 若发现必须修改“禁止修改”范围，必须停止并说明缺失的公共能力、建议新增的接口及影响范围，等待底座负责人确认后再继续，不能自行复制或绕开底座逻辑。

17. 实施顺序

1. 建立配置、错误、日志、数据库会话和公共数据契约。
2. 实现 AgentRegistry、AgentFactory 和最小 BaseAgent。
3. 接入工具执行器、模型网关和合规服务。
4. 实现短期会话、请求幂等、对话审计和通用 Outbox，打通最小客服 Agent。
5. 实现配置发布中心、配置快照和回滚。
6. 实现多模型路由、健康检查、备用链和 Prompt 版本管理。
7. 实现中长期记忆、证据冲突、Outbox 和多存储同步。
8. 实现 Neo4j 图投影、受控关系视图和 MySQL 权威校验。
9. 接入知识库、MySQL 降级检索和知识审核状态。
10. 接入其他业务 Agent，并完成跨 Agent 契约测试。
11. 完成压测、安全测试、迁移演练和故障恢复演练。

18. 完成标准

• 四种 Agent 可由工厂按权限创建。
• 公共七步骨架不可被子类绕过。
• 工具调用、模型输出、引用、对话和审计可通过同一 trace_id 追踪。
• 任一外部依赖故障均有明确降级或阻断策略。
• 跨客户数据访问为零，敏感字段泄漏为零。
• 数据库迁移可从空库完整执行，并可在测试环境完成升级和回滚演练。
• 组员新增业务 Agent 时不需要修改长期记忆、Neo4j、配置中心或模型路由实现。
• 配置、Prompt、模型、工具、记忆和关系视图版本均可通过 trace_id 还原。
• 原有数据库表名和已有字段定义在所有迁移中保持不变。

19. 实现现状与设计差异

本节记录设计与实现已经分叉的地方，避免开发者按本文档早期描述写出与代码不符的调用。 核对对象：app/core/contracts.py、app/service/agent/base.py、app/service/agent/factory.py、 app/service/agent/governance.py、app/worker/runtime.py、app/api/ 与 app/main.py 的实际结构。

两者冲突时以代码为准，并在本节补充说明；不要为了迁就早期描述去改这些已经过端到端验收的结构。

19.1 工程目录

本文档 §4 描述	代码实际
app/controller	app/api/controllers
app/view	app/api/views
service/agent/model_gateway/、service/tool/、service/memory/ 等子包	未按子包拆分，model_gateway.py、tool_executor.py、memory_service.py 为平铺模块
未描述	新增 app/api/schemas（HTTP DTO）、app/api/dependencies（Session 与鉴权依赖）、app/worker（独立 Worker 进程）、app/infrastructure（Redis/Milvus/Neo4j/行情适配器）

移动源码会破坏既有测试与导入路径，因此以代码实际目录为准，不回改。

19.2 核心数据契约

契约	早期设计描述	实现
AgentRequest	含 end_session；幂等键 8–64；metadata 用受约束字面量	无 end_session；幂等键 16–128 且仅字母数字与 -/_；metadata 只有 locale/client_version/ui_entry 三个可选字符串
RequestContext	frozen dataclass；user_id: int；roles 为 frozenset；含 assigned_customer_ids	Pydantic frozen model；user_id: str；roles/customer_ids 为 tuple；数据范围经 permissions + permission_scopes 表达
IntentResult	字段名 name；confidence 为 Decimal；含 entities	字段名 intent；confidence: float（0–1）；另有 needs_clarification；无 entities
CoreResult	direct_reply 与 generation_material 二选一	单一 text，另含 intent/source_references/tool_calls/transfer_required
AgentResult	含 degraded/degradation_reason	run_id + result: CoreResult + usage
RunProgressEvent	event/trace_id/data	event_type/run_id/payload
AgentDefinition	含 display_name、意图阈值、合规与记忆策略等	仅 agent_type/version/allowed_tools/allowed_roles/allowed_portals/supported_intents；其余能力在治理层与发布配置中表达

精度取舍：confidence 由 Decimal 改为 float 会失去十进制精度契约，写入 conversation_message.confidence（DECIMAL(5,4)）时由持久化层转回 Decimal。

19.3 BaseAgent 执行骨架

• 早期描述要求 execute() 内完成用户消息落库、complete_run() 与分块 delta 事件。 实现中 execute() 只产出 start 与 done 两个事件；落库与终态由 Worker 侧 AgentPersistenceService.complete_run() 在同一事务内完成，delta/replace 由 SSE View 从数据库结果重建（即结果级恢复，不做事件级续传）。
• 输入侧合规在实现中合并为事后 AgentGovernance.review()（引用校验、禁止表达、 号码脱敏），没有独立的输入阶段合规方法。
• 七步骨架的强制手段是 BaseAgent.__init_subclass__ 的禁用名单（子类覆写 execute/validate_access/resolve_config/recall_memory/call_tool 等会直接抛 TypeError），而不是靠约定。

19.4 其它已确认差异

• 模型、记忆、合规等能力在实现中收敛为「治理聚合」：AgentGovernance 协议 + BaseAgent.bind_governance/bind_model_service/bind_tool_executor/bind_intent_classifier 注入，而不是各自独立的 Protocol 实现类。
• SSE 按结果级恢复实现：delta 载荷键为 content，并带一个本设计未定义的 replay 开关 （终态重连时改发 replace）；id 形如 {run_id}:{event}:{index}。
• 错误模型为 AgentError 子类携带 code/message/status_code，由 app/main.py 的 异常处理器统一封装成 {error:{code,message,retryable,field_errors}, meta:{trace_id}}。
• run_id 与 trace_id 已分离：trace_id 由请求入口生成，run_id 是运行实体主键。