repo_description: LLM驱动的 A/H/美股智能分析器：多数据源行情 + 实时新闻 + LLM决策仪表盘 + 多渠道推送，零成本定时运行，纯白嫖. LLM-powered stock analysis system for A/H/US markets.
repo_topics: agent ai aigc gemini llm quant quantitative-trading rag stock
code: # -*- coding: utf-8 -*-
  """
  Shared factory for building fully-configured AgentExecutor instances.
  
  Centralises construction to eliminate boilerplate duplicated across
  api/v1/endpoints/agent.py, bot/commands/chat.py, bot/commands/ask.py,
  and src/core/pipeline.py.
  
  Performance notes
  -----------------
  * ``ToolRegistry`` is built once and cached at module level — tool
    registrations are immutable after setup so the object is safe to share
    across every request.
  * ``SkillManager`` is expensive to create (loads YAML files from disk).
    A prototype is built on first use and cheap ``deepcopy`` clones are
    returned for each request, preserving thread-safety (``activate()``
    mutates internal state).
  
  Usage::
  
      from src.agent.factory import build_agent_executor
  
      executor = build_agent_executor(config, skills=["bull_trend", "shrink_pullback"])
      result   = executor.chat(message="...", session_id="...")
  """
  
  import copy
  import logging
  from dataclasses import dataclass
  from typing import List, Optional
  
  from src.config import AGENT_MAX_STEPS_DEFAULT
  
  logger = logging.getLogger(__name__)
  
  # ---------------------------------------------------------------------------
  # Module-level caches
  # ---------------------------------------------------------------------------
  _TOOL_REGISTRY = None
  _SKILL_MANAGER_PROTOTYPE = None
  # Sentinel used as initial value so None (i.e. no custom dir) compares as "changed"
  # on the very first call, forcing a build rather than accidentally skipping it.
  _SENTINEL = object()
  # Track which custom_dir the prototype was built with so we can invalidate
  # the cache if AGENT_SKILL_DIR changes at runtime (e.g. via config reload).
  _SKILL_MANAGER_CUSTOM_DIR: object = _SENTINEL
  
  
  @dataclass
  class SkillPromptState:
      """Resolved skill activation + prompt fragments for analysis entrypoints."""
  
      skill_manager: object
      skills_to_activate: List[str]
      explicit_skill_selection: bool
      use_legacy_default_prompt: bool
      skill_instructions: str
      default_skill_policy: str
      technical_skill_policy: str
  
  
  def _normalize_skill_ids(
      skill_ids: Optional[List[str]],
      *,
      available_skill_ids: set[str],
  ) -> tuple[List[str], List[str]]:
      """Return validated skill ids plus unknown ids, preserving input order."""
      normalized: List[str] = []
      unknown: List[str] = []
  
      for skill_id in skill_ids or []:
          if not isinstance(skill_id, str):
              continue
          cleaned = skill_id.strip()
          if not cleaned:
              continue
          if cleaned == "all":
              if "all" not in normalized:
                  normalized.append("all")
              continue
          if cleaned in available_skill_ids:
              if cleaned not in normalized:
                  normalized.append(cleaned)
              continue
          if cleaned not in unknown:
              unknown.append(cleaned)
  
      return normalized, unknown
  
  
  def _resolve_selected_skill_ids(
      *,
      requested_skills: Optional[List[str]],
      configured_skills: Optional[List[str]],
      default_skills: List[str],
      available_skill_ids: set[str],
  ) -> tuple[List[str], bool]:
      """Resolve active skill ids and whether they came from a valid explicit selection."""
      selection_source = None
      raw_skill_ids = None
      if requested_skills is not None:
          selection_source = "request"
          raw_skill_ids = requested_skills
      elif configured_skills is not None:
          selection_source = "config"
          raw_skill_ids = configured_skills
      else:
          return list(default_skills), False
  
      selected_skill_ids, unknown_skill_ids = _normalize_skill_ids(
          raw_skill_ids,
          available_skill_ids=available_skill_ids,
      )
      if unknown_skill_ids:
          logger.warning(
              "[AgentFactory] Ignoring unknown %s skill ids: %s",
              selection_source,
              unknown_skill_ids,
          )
      if selected_skill_ids:
          return selected_skill_ids, True
  
      if raw_skill_ids:
          logger.warning(
              "[AgentFactory] No valid %s skills remain after validation; falling back to default skills: %s",
              selection_source,
              default_skills,
          )
      return list(default_skills), False
  
  
  def _should_use_legacy_default_prompt(
      *,
      skills_to_activate: List[str],
      explicit_skill_selection: bool,
      skill_catalog: List[object],
  ) -> bool:
      """Keep the legacy prompt only for the implicit built-in bull_trend fallback."""
      if explicit_skill_selection or skills_to_activate != ["bull_trend"]:
          return False
  
      bull_trend_skill = next(
          (
              skill
              for skill in skill_catalog
              if str(getattr(skill, "name", "")).strip() == "bull_trend"
          ),
          None,
      )
      return getattr(bull_trend_skill, "source", None) == "builtin"
  
  
  def get_tool_registry():
      """Return a cached ToolRegistry (built once, shared across requests)."""
      global _TOOL_REGISTRY
      if _TOOL_REGISTRY is not None:
          return _TOOL_REGISTRY
  
      from src.agent.tools.registry import ToolRegistry
      from src.agent.tools.data_tools import ALL_DATA_TOOLS
      from src.agent.tools.analysis_tools import ALL_ANALYSIS_TOOLS
      from src.agent.tools.search_tools import ALL_SEARCH_TOOLS
      from src.agent.tools.market_tools import ALL_MARKET_TOOLS
      from src.agent.tools.backtest_tools import ALL_BACKTEST_TOOLS
  
      registry = ToolRegistry()
      for tool_fn in ALL_DATA_TOOLS + ALL_ANALYSIS_TOOLS + ALL_SEARCH_TOOLS + ALL_MARKET_TOOLS + ALL_BACKTEST_TOOLS:
          registry.register(tool_fn)
  
      _TOOL_REGISTRY = registry
      logger.info("[AgentFactory] ToolRegistry cached (%d tools)", len(registry._tools) if hasattr(registry, "_tools") else -1)
      return _TOOL_REGISTRY
  
  
  def get_skill_manager(config=None):
      """Return a deepcopy-clone of the cached SkillManager prototype.
  
      The prototype is initialised from disk on first call; subsequent calls
      return ``copy.deepcopy(prototype)`` which is ~10× faster than re-reading
      YAML files.  Each clone is independent so ``.activate()`` calls do not
      bleed between requests.
  
      Cache invalidation: if ``config.agent_skill_dir`` changes at runtime
      (e.g. via the web settings reload), the prototype is rebuilt automatically.
      """
      global _SKILL_MANAGER_PROTOTYPE, _SKILL_MANAGER_CUSTOM_DIR
  
      if config is None:
          from src.config import get_config
          config = get_config()
  
      current_custom_dir = getattr(config, "agent_skill_dir", None)
      if _SKILL_MANAGER_PROTOTYPE is not None and current_custom_dir == _SKILL_MANAGER_CUSTOM_DIR:
          return copy.deepcopy(_SKILL_MANAGER_PROTOTYPE)
  
      from src.agent.skills.base import SkillManager
  
      if _SKILL_MANAGER_PROTOTYPE is not None:
          logger.info("[AgentFactory] SkillManager prototype invalidated (agent_skill_dir changed: %r -> %r)",
                      _SKILL_MANAGER_CUSTOM_DIR, current_custom_dir)
  
      skill_manager = SkillManager()
      skill_manager.load_builtin_skills()
  
      if current_custom_dir:
          try:
              skill_manager.load_custom_skills(current_custom_dir)
          except Exception as exc:
              logger.warning("[AgentFactory] Failed to load custom skills from %s: %s", current_custom_dir, exc)
  
      _SKILL_MANAGER_PROTOTYPE = skill_manager
      _SKILL_MANAGER_CUSTOM_DIR = current_custom_dir
      logger.info("[AgentFactory] SkillManager prototype cached (%d skills)", len(skill_manager._skills))
      return copy.deepcopy(_SKILL_MANAGER_PROTOTYPE)
  
  
  def resolve_skill_prompt_state(config=None, skills: Optional[List[str]] = None) -> SkillPromptState:
      """Resolve active skills and prompt fragments for analyzer / agent entrypoints."""
      if config is None:
          from src.config import get_config
          config = get_config()
  
      from src.agent.skills.defaults import (
          get_default_active_skill_ids,
          get_default_technical_skill_policy,
          get_default_trading_skill_policy,
      )
  
      skill_manager = get_skill_manager(config)
      skill_catalog = list(skill_manager.list_skills())
      available_skill_ids = {
          str(getattr(skill, "name", "")).strip()
          for skill in skill_catalog
          if str(getattr(skill, "name", "")).strip()
      }
      configured_skills = getattr(config, "agent_skills", None)
      if configured_skills == []:
          configured_skills = None
      default_skills = get_default_active_skill_ids(
          skill_catalog,
          available_skill_ids=available_skill_ids or None,
      )
      skills_to_activate, explicit_skill_selection = _resolve_selected_skill_ids(
          requested_skills=skills,
          configured_skills=configured_skills,
          default_skills=default_skills,
          available_skill_ids=available_skill_ids,
      )
  
      use_legacy_default_prompt = _should_use_legacy_default_prompt(
          skills_to_activate=skills_to_activate,
          explicit_skill_selection=explicit_skill_selection,
          skill_catalog=skill_catalog,
      )
  
      skill_manager.activate(skills_to_activate)
      logger.info("[AgentFactory] Activated skills: %s", skills_to_activate)
  
      return SkillPromptState(
          skill_manager=skill_manager,
          skills_to_activate=skills_to_activate,
          explicit_skill_selection=explicit_skill_selection,
          use_legacy_default_prompt=use_legacy_default_prompt,
          skill_instructions=skill_manager.get_skill_instructions(),
          default_skill_policy=get_default_trading_skill_policy(
              explicit_skill_selection=not use_legacy_default_prompt,
          ),
          technical_skill_policy=get_default_technical_skill_policy(
              explicit_skill_selection=not use_legacy_default_prompt,
          ),
      )
  
  
  def build_agent_executor(config=None, skills: Optional[List[str]] = None):
      """Build and return a configured AgentExecutor (or future orchestrator).
  
      When ``AGENT_ARCH=multi``, this returns an orchestrator that manages
      multiple specialised agents. Otherwise it returns the legacy single-agent
      executor.
  
      Args:
          config: Application config object.  When *None*, ``get_config()`` is
                  called automatically.
          skills: Skill ids to activate.  When *None* falls back to
                  ``config.agent_skills``; if that is also empty falls back to
                  the central default skill set.
  
      Returns:
          A ready-to-call :class:`src.agent.executor.AgentExecutor` instance.
      """
      if config is None:
          from src.config import get_config
          config = get_config()
  
      arch = getattr(config, "agent_arch", "single")
  
      from src.agent.llm_adapter import LLMToolAdapter
  
      registry = get_tool_registry()
      prompt_state = resolve_skill_prompt_state(config, skills=skills)
      skill_manager = prompt_state.skill_manager
      logger.info(
          "[AgentFactory] Resolved skill prompt state: skills=%s (arch=%s, explicit=%s, legacy_default_prompt=%s)",
          prompt_state.skills_to_activate,
          arch,
          prompt_state.explicit_skill_selection,
          prompt_state.use_legacy_default_prompt,
      )
  
      llm_adapter = LLMToolAdapter(config)
  
      if arch == "multi":
          return _build_orchestrator(
              config,
              registry,
              llm_adapter,
              skill_manager,
              technical_skill_policy=prompt_state.technical_skill_policy,
          )
  
      from src.agent.executor import AgentExecutor
      return AgentExecutor(
          tool_registry=registry,
          llm_adapter=llm_adapter,
          skill_instructions=prompt_state.skill_instructions,
          default_skill_policy=prompt_state.default_skill_policy,
          use_legacy_default_prompt=prompt_state.use_legacy_default_prompt,
          max_steps=getattr(config, "agent_max_steps", AGENT_MAX_STEPS_DEFAULT),
          timeout_seconds=getattr(config, "agent_orchestrator_timeout_s", 0),
      )
  
  
  def _build_orchestrator(config, registry, llm_adapter, skill_manager, *, technical_skill_policy: str = ""):
      """Build and return an :class:`AgentOrchestrator` (multi-agent mode).
  
      The orchestrator presents the same ``run()`` / ``chat()`` interface as
      :class:`AgentExecutor` so callers need no changes.
      """
      from src.agent.orchestrator import AgentOrchestrator
  
      mode = getattr(config, "agent_orchestrator_mode", "standard")
      logger.info("[AgentFactory] Building AgentOrchestrator (mode=%s)", mode)
  
      return AgentOrchestrator(
          tool_registry=registry,
          llm_adapter=llm_adapter,
          skill_instructions=skill_manager.get_skill_instructions(),
          technical_skill_policy=technical_skill_policy,
          max_steps=getattr(config, "agent_max_steps", AGENT_MAX_STEPS_DEFAULT),
          mode=mode,
          skill_manager=skill_manager,
          config=config,
      )
  
  
  # Keep legacy alias so any external callers using the old name still work.
  build_executor = build_agent_executor
  