group_fqcd_jr/app/service/agent/bootstrap.py

import logging
from functools import lru_cache
from typing import Any, cast

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import get_settings
from app.core.errors import RecoverableAgentError
from app.core.fund_contracts import FundQuoteQuery
from app.core.knowledge_contracts import KnowledgeSearchInput
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.factory import AgentFactory
from app.service.agent.governance import PlatformGovernance
from app.service.agent.implementations.customer_service import CustomerServiceAgent
from app.service.agent.implementations.fund_query_demo import FundQueryDemoAgent
from app.service.fund_quote_service import query_fund_quote_tool
from app.service.intent_classifier import IntentClassifier
from app.service.knowledge_search_service import KnowledgeSearchService
from app.service.knowledge_tool import knowledge_search_tool
from app.service.memory_recall_service import MemoryRecallService
from app.service.model_gateway import (
    DatabaseModelEndpointResolver,
    DatabaseModelGateway,
    ModelDispatchService,
    ModelEmbeddingService,
    ModelGenerationService,
)
from app.service.runtime_config_service import load_active_intent_configs
from app.service.suitability_service import SuitabilityToolInput, suitability_tool_handler
from app.service.tool_executor import ToolDefinition, ToolExecutor, ToolRegistry

logger = logging.getLogger(__name__)


@lru_cache(maxsize=1)
def get_model_service() -> ModelGenerationService:
    """生产模型装配的唯一入口；业务 Agent 与 Worker 记忆抽取共用同一实例。"""
    return ModelGenerationService(ModelDispatchService(DatabaseModelGateway()))


@lru_cache(maxsize=1)
def get_memory_cache_adapter() -> MemoryCacheAdapter | None:
    """Redis 记忆缓存适配器；构造失败返回 None——缓存只是优化层，不得阻塞召回。"""
    try:
        from redis.asyncio import Redis

        settings = get_settings()
        client = Redis.from_url(
            settings.redis_url,
            socket_connect_timeout=settings.redis_connect_timeout_seconds,
            socket_timeout=settings.redis_connect_timeout_seconds,
            decode_responses=True,
        )
        return MemoryCacheAdapter(client)
    except Exception:
        logger.warning("memory cache adapter unavailable; recall runs without cache",
                       exc_info=True)
        return None


@lru_cache(maxsize=1)
def get_fund_quote_cache() -> FundQuoteCache | None:
    """行情短缓存：与记忆召回共用同一个 Redis 适配器。

    Redis 不可用时这里仍会返回适配器，但读写异常由 `MemoryCacheAdapter` 内部
    吞掉并以 `degraded` 语义返回，行情查询会退化为直连外部数据源而不会阻塞；
    适配器构造失败（例如缺少 redis 依赖）则返回 None，效果相同。缓存永远只是
    优化层，不得让行情查询因缓存故障失败。
    """
    adapter = get_memory_cache_adapter()
    if adapter is None:
        return None
    return FundQuoteCache(adapter)


@lru_cache(maxsize=1)
def get_vector_memory_adapter() -> VectorMemoryAdapter | None:
    """Milvus 适配器；构造失败返回 None（语义通道关闭），不影响结构化召回。"""
    try:
        from pymilvus import MilvusClient  # type: ignore[import-untyped]

        settings = get_settings()
        client = MilvusClient(uri=settings.milvus_uri, token=settings.milvus_token or None)
        return VectorMemoryAdapter(client, settings.milvus_collection)
    except Exception:
        logger.warning("vector memory adapter unavailable; semantic recall disabled",
                       exc_info=True)
        return None


@lru_cache(maxsize=1)
def get_memory_embedding_service() -> ModelEmbeddingService:
    """文本向量化入口：与文本生成共用同一套端点解析与受控降级。"""
    return ModelEmbeddingService(ModelDispatchService(DatabaseModelGateway()))


async def _embed_text(text: str) -> list[float]:
    """把文本向量化；端点来自发布配置（task_type=embedding），无端点时失败关闭。"""
    endpoints = await DatabaseModelEndpointResolver().resolve(
        agent_type="memory_recall", task_type="embedding"
    )
    if not endpoints:
        raise RecoverableAgentError("没有可用的 embedding 端点")
    execution = await get_memory_embedding_service().embed(endpoints, text)
    return execution.vector


@lru_cache(maxsize=1)
def get_knowledge_search_service() -> KnowledgeSearchService:
    """客服知识检索装配：Milvus 客户端 + 向量化入口。

    与记忆的语义通道同一取向：Milvus 不可达或缺少 embedding 端点时**不抛异常**，
    而是返回 `available=False` 的实例，检索结果标记 `degraded`，由客服 Agent 走
    「引导客户致电人工客服」。基础设施故障不该表现成客户可见的错误。
    """
    client = None
    try:
        from pymilvus import MilvusClient

        settings = get_settings()
        client = MilvusClient(uri=settings.milvus_uri, token=settings.milvus_token or None)
    except Exception:
        logger.warning("knowledge vector client unavailable; search degrades", exc_info=True)
    return KnowledgeSearchService(client, _embed_text)


def build_memory_recall_service(session: AsyncSession) -> MemoryRecallService:
    """记忆召回组装：结构化召回始终可用，Redis 缓存与语义通道可用时叠加。

    语义通道需要**两件事同时具备**：可达的 Milvus 与可用的 embedding 端点。缺少
    embedding 端点时向量化按设计失败关闭，结果标记 `embedding_failed` 并保留结构化
    召回——这是配置缺口而非功能缺失，配置端点后语义召回无需改代码即可生效。
    """
    vector = get_vector_memory_adapter()
    return MemoryRecallService(
        session,
        vector=vector,
        cache=get_memory_cache_adapter(),
        embed=_embed_text if vector is not None else None,
    )


@lru_cache(maxsize=1)
def get_agent_factory() -> AgentFactory:
    """HTTP 与 Worker 共用的唯一底座依赖组装入口。"""
    registry = ToolRegistry()
    registry.register(ToolDefinition(
        name="check_suitability",
        input_model=SuitabilityToolInput,
        handler=cast(Any, suitability_tool_handler),
        required_permission="suitability:read",
        allowed_roles=("customer", "advisor", "operator", "admin"),
    ))
    registry.register(ToolDefinition(
        name="query_fund_quote",
        input_model=FundQuoteQuery,
        handler=cast(Any, query_fund_quote_tool),
        required_permission="fund:quote:read",
        allowed_roles=("customer", "advisor", "operator", "risk_operator", "admin"),
        # C3 数值依据：EastmoneyAdapterFactory 的单代码最坏预算为
        # 4.0s(单次) + 0.2s(退避) + 4.0s(重试) = 8.2s，且 FundQuoteService
        # 已按代码并发，总耗时与代码数量无关。15s ≈ 8.2s × 1.8，余量覆盖
        # DNS/TLS 建连与事件循环调度开销。原值 5s 小于适配器默认单次超时
        # （12s）与重试预算，多代码查询必然先撞工具超时。
        timeout_seconds=15,
    ))
    registry.register(ToolDefinition(
        name="search_knowledge",
        input_model=KnowledgeSearchInput,
        handler=cast(Any, knowledge_search_tool),
        # 复用既有权限码（customer 角色已具备），不新增权限点
        required_permission="knowledge:reference:read",
        allowed_roles=("customer", "advisor", "operator", "admin"),
        # 检索含一次 embedding 调用 + 三次集合检索；embedding 实测 0.42s，
        # 10s 覆盖冷启动与 Milvus 抖动，又不至于让客户等太久
        timeout_seconds=10,
    ))
    model_service = get_model_service()
    endpoint_resolver = DatabaseModelEndpointResolver()
    factory = AgentFactory(
        # 记忆召回接入统一治理链：Agent 的 recall_memory 走组合召回服务，
        # 而不是每个 Agent 自行决定召回方式。
        governance=PlatformGovernance(recall_factory=build_memory_recall_service),
        model_service=model_service,
        tool_executor=ToolExecutor(registry),
        intent_classifier=IntentClassifier(
            model_service, config_loader=load_active_intent_configs
        ),
        intent_endpoint_resolver=endpoint_resolver,
    )
    register_business_agents(factory)
    return factory


def register_business_agents(factory: AgentFactory) -> None:
    """业务 Agent 的统一注册入口：组员在这里登记自己的一行 `factory.register(...)`。

    HTTP 服务与 Worker 共用 `get_agent_factory()` 返回的同一个工厂，因此这里注册
    一次即可在两个入口生效。注册只声明"代码允许什么"：真正能调用哪些工具，还要
    看当前 active 的 `config_release` 里为该 `agent_type:intent` 发布的工具白名单
    （两者取交集，缺发布配置时白名单为空、工具失败关闭）。
    """
    factory.register(
        FundQueryDemoAgent.definition,
        lambda _context: FundQueryDemoAgent(FundQueryDemoAgent.definition),
    )
    # 客服 Agent：只回答能溯源到公司资料的问题，答不了引导客户致电人工客服。
    # 它声明了 5 个意图，但真正能调用 search_knowledge 的范围由发布配置逐意图收窄
    # （`agent_tools` 里 `customer_service:<intent>`），未发布的意图工具失败关闭。
    factory.register(
        CustomerServiceAgent.definition,
        lambda _context: CustomerServiceAgent(CustomerServiceAgent.definition),
    )