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 KnowledgeQuery 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 from app.service.agent.offsite_fund_agent import OffsiteFundAgent from app.service.financial_nl2sql_service import query_financial_data_tool from app.service.fund_quote_service import query_fund_quote_tool from app.service.intent_classifier import IntentClassifier from app.service.knowledge_tool_service import query_knowledge_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.resolved_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 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="query_financial_data", input_model=FinancialNL2SQLInput, handler=cast(Any, query_financial_data_tool), required_permission="financial:nl2sql:read", allowed_roles=("advisor", "operator", "admin", "super_admin"), timeout_seconds=10, )) registry.register(ToolDefinition( name="query_knowledge", input_model=KnowledgeQuery, handler=cast(Any, query_knowledge_tool), required_permission="knowledge:query", allowed_roles=("visitor", "customer"), # 15 秒向量端点预算外预留检索和权威回查时间，防止正常降级被工具层提前中断。 timeout_seconds=20, )) 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), ) factory.register( OffsiteFundAgent.definition, lambda _context: OffsiteFundAgent(OffsiteFundAgent.definition), ) factory.register( CustomerServiceAgent.definition, lambda _context: CustomerServiceAgent(), )