import json import logging from collections.abc import AsyncGenerator from dataclasses import dataclass from pathlib import Path from app.engine.providers.base import AIProvider from app.engine.providers.glm import GLMProvider from app.engine.config import settings from app.engine.services.boundary_filter import DEFAULT_WALL from app.engine.wall_schema import CompiledWall logger = logging.getLogger(__name__) def _format_messages(messages: list[dict]) -> str: lines = [] for m in messages: role = m.get("role", "unknown") content = m.get("content", "") label = {"user": "客户", "assistant": "顾问"}.get(role, role) lines.append(f"{label}: {content}") return "\n".join(lines) # 话术池注入的语言行序(确定性输出;未列语言按字典序补位,不丢池) _DEFLECTION_LANG_ORDER = ("en", "zh", "es") def _format_deflection_pool(deflections) -> str: """wall.json deflections 池 → system.md $DEFLECTION_POOL 注入块。 按语言分组一行("- EN: …" / "- ZH: …"),条目引号包裹 " / " 分隔—— 紧凑不膨胀(设计 R1: 话术单一事实源,提示词侧只引用不复制)。 只渲染主池(deflections);deflections_contact 是已留资态的墙替换池, 本就不属于 system.md 教学内容,不注入。 已知边界(审查 Minor-1 注记): 条目自身含双引号/换行时暂不转义—— 现网池线均为单行纯文本,Phase 2 后台编辑器批次统一处理(输入校验+转义)。 """ pools = dict(deflections or {}) langs = [l for l in _DEFLECTION_LANG_ORDER if pools.get(l)] langs += sorted(l for l in pools if l not in _DEFLECTION_LANG_ORDER and pools.get(l)) return "\n".join( f"- {lang.upper()}: " + " / ".join(f'"{line}"' for line in pools[lang]) for lang in langs ) @dataclass(frozen=True) class PromptProvider: """租户提示词四件套持有者(Task 6, R1 话术单一事实源)。 frozen(Task 9 收口, Task 6 审查遗留): 实例跨租户运行时/模块级 DEFAULT_PROVIDER 共享, 不可变防共享单例被毒化。 - from_dir(dir): 读 system/company_facts/extractor/evaluator 四个 md, 存实例字段——提示词属于租户(seed_profiles//prompts/),不再有 代码级 app/engine/prompts 副本; - build_system(wall_deflections): system.md 的 $DEFLECTION_POOL 占位符 替换为按语言枚举的话术清单 + "\\n\\n" + company_facts(行为 prompt 在前、事实卡居中、当轮提醒最后拼接——拼接位在 AIService 内); - build_extractor/build_evaluator: $existing_info/$messages/$profile/ $recent_messages 替换逻辑自 AIService 原样搬入(行为等价)。 """ system: str company_facts: str extractor: str evaluator: str @classmethod def from_dir(cls, prompts_dir) -> "PromptProvider": d = Path(prompts_dir) return cls( system=(d / "system.md").read_text(encoding="utf-8"), company_facts=(d / "company_facts.md").read_text(encoding="utf-8"), extractor=(d / "extractor.md").read_text(encoding="utf-8"), evaluator=(d / "evaluator.md").read_text(encoding="utf-8"), ) def build_system(self, wall_deflections=None) -> str: system = self.system if "$DEFLECTION_POOL" in system: system = system.replace( "$DEFLECTION_POOL", _format_deflection_pool(wall_deflections)) return system + "\n\n" + self.company_facts def build_extractor(self, existing_info: dict, messages: list[dict]) -> str: return self.extractor.replace( "$existing_info", json.dumps(existing_info or {}, ensure_ascii=False) ).replace("$messages", _format_messages(messages)) def build_evaluator(self, profile: dict, recent_messages: list[dict]) -> str: return self.evaluator.replace( "$profile", json.dumps(profile, ensure_ascii=False) ).replace("$recent_messages", _format_messages(recent_messages[-6:])) # 代码级默认 = jiancai 出厂(路径锚定模块位置,与 cwd 无关): 旧测试与 # 无租户上下文的默认路径用;租户路径一律经 TenantRuntime.prompts 注入。 DEFAULT_PROVIDER = PromptProvider.from_dir( Path(__file__).resolve().parents[3] / "seed_profiles" / "jiancai" / "prompts") # model 的出厂默认值——视同"未设置"哨兵: 租户未定制(merged 值 == 出厂默认) # 时回退 env AI_MODEL,部署级模型选择不因接线失效。显式配了其他值则用租户的。 _FACTORY_DEFAULT_MODEL = "glm-4-flash" def resolve_ai_settings(config_data: dict | None) -> dict: """租户 config.ai → provider 实例化参数(Task 9 接线; env 回退)。 api_key/base_url 空槽(None/空串)→ env 智谱凭据(平台默认集团 key, spec §5.5 "用平台默认"状态下租户侧不存密钥);model 空或等于出厂默认 → env AI_MODEL。 已知边界: provider 出厂 "glm" 同样遮 env AI_PROVIDER——当前单 provider 无实害,多 provider 接入时再改哨兵结构(与 push 数值同款纪律)。 config_data=None 路径仅供测试直调,生产调用一律传 runtime.config.data。 """ ai = ((config_data or {}).get("ai")) or {} model = ai.get("model") if not model or model == _FACTORY_DEFAULT_MODEL: model = settings.AI_MODEL or _FACTORY_DEFAULT_MODEL return { "provider": ai.get("provider") or settings.AI_PROVIDER, "api_key": ai.get("api_key") or settings.ZHIPU_API_KEY, "base_url": ai.get("base_url") or settings.ZHIPU_BASE_URL, "model": model, } def create_provider(config_data: dict | None = None) -> AIProvider: """provider 工厂(Task 9: 按租户配置实例化 glm;None 走 env 平台默认)。""" cfg = resolve_ai_settings(config_data) if cfg["provider"] == "glm": return GLMProvider( api_key=cfg["api_key"], model=cfg["model"], base_url=cfg["base_url"], ) raise ValueError(f"Unknown AI provider: {cfg['provider']}") class AIService: def __init__(self, provider: AIProvider | None = None, prompts: PromptProvider | None = None, wall: CompiledWall | None = None, config_data: dict | None = None): self.provider = provider or create_provider(config_data) # Task 6: 提示词经 PromptProvider 注入(租户四件套);默认 = jiancai # 出厂 + 出厂墙话术池(与 DEFAULT_PROVIDER 同源,单一事实源 wall.json) self.prompts = prompts or DEFAULT_PROVIDER # Task 9 收口: wall 参数定型为 CompiledWall(话术池只取 .deflections, # 与 prompts 同为租户运行时产物);缺省 = jiancai 出厂墙 self.wall_deflections = (wall.deflections if wall is not None else DEFAULT_WALL.deflections) # 对话基座: 行为 prompt 在前(红线显著性),事实卡居中(参考资料), # 当轮 context 提醒最后拼接(在 chat/stream_chat 内,保持最新鲜) self._base_chat_prompt = self.prompts.build_system(self.wall_deflections) async def chat(self, messages: list[dict], context: str = "") -> str: """Main conversation: get AI reply to user message.""" prompt = self._base_chat_prompt + ("\n\n" + context if context else "") return await self.provider.chat(messages, prompt) async def stream_chat(self, messages: list[dict], context: str = "") -> AsyncGenerator[str, None]: """Stream AI reply token by token.""" prompt = self._base_chat_prompt + ("\n\n" + context if context else "") async for token in self.provider.stream_chat(messages, prompt): yield token async def extract_info( self, messages: list[dict], existing_info: dict ) -> dict: """Extract structured info from conversation history.""" prompt = self.prompts.build_extractor(existing_info, messages) _, extracted = await self.provider.chat_with_extraction( messages=[], system_prompt=prompt ) return extracted async def evaluate( self, profile: dict, recent_messages: list[dict] ) -> dict: """Evaluate customer intent level.""" prompt = self.prompts.build_evaluator(profile, recent_messages) _, result = await self.provider.chat_with_extraction( messages=[], system_prompt=prompt ) return self._normalize_evaluation(result) @staticmethod def _normalize_evaluation(result: dict) -> dict: intent = result.get("intent_level", "pending") if intent not in ("interested", "not_interested", "pending"): intent = "pending" return { "intent_level": intent, "reasoning": result.get("reasoning", ""), "contact_info": result.get("contact_info", {}), "info_sufficient": result.get("info_sufficient", False), }