"""供 CLI 和外部嵌入使用的 LLM client 工厂。""" from __future__ import annotations import os import logging from pam_deploy_graph.logging_utils import json_for_log from .base import LlmClient from .openai_compatible import OpenAICompatibleLlmClient, load_prompt_text from .rule_based import RuleBasedLlmClient logger = logging.getLogger(__name__) def build_llm_client( *, base_url: str | None = None, api_key: str | None = None, model: str | None = None, action_analysis_prompt_path: str | None = None, ) -> LlmClient: """根据显式参数或环境变量构造 LLM client。""" actual_base_url = base_url if base_url is not None else os.getenv("PAM_LLM_BASE_URL", "") actual_api_key = api_key if api_key is not None else os.getenv("PAM_LLM_API_KEY", "") actual_model = model if model is not None else os.getenv("PAM_LLM_MODEL", "") actual_action_prompt_path = ( action_analysis_prompt_path if action_analysis_prompt_path is not None else os.getenv("PAM_LLM_ACTION_ANALYSIS_PROMPT_FILE", "") ) logger.info( "构建 LLM client base_url=%s model=%s has_api_key=%s action_prompt_path=%s explicit=%s", actual_base_url, actual_model, bool(actual_api_key), actual_action_prompt_path, json_for_log( { "base_url": base_url, "api_key": api_key, "model": model, "action_analysis_prompt_path": action_analysis_prompt_path, } ), ) if not actual_base_url and not actual_api_key and not actual_model: logger.info("未配置真实 LLM,使用 RuleBasedLlmClient fallback") return RuleBasedLlmClient() missing = [] if not actual_base_url: missing.append("base_url") if not actual_model: missing.append("model") if missing: logger.info("LLM 配置不完整 missing=%s", missing) raise ValueError(f"LLM 配置不完整,缺少: {', '.join(missing)}") client = OpenAICompatibleLlmClient( base_url=actual_base_url, api_key=actual_api_key, model=actual_model, action_analysis_prompt=load_prompt_text(actual_action_prompt_path), ) logger.info("真实 LLM client 构建完成 client=%s model=%s has_api_key=%s", type(client).__name__, actual_model, bool(actual_api_key)) return client