160 lines
5.0 KiB
Python
160 lines
5.0 KiB
Python
"""
|
||
爽文开书优化（fiction.open_book）— LLM 调用逻辑。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from langchain_core.messages import HumanMessage, SystemMessage
|
||
|
||
from models.fiction_models import (
|
||
EmotionFlow,
|
||
FictionGuideWorldbook,
|
||
OpenBookResult,
|
||
)
|
||
from services.fiction_prompt_utils import resolve_prompt
|
||
from services.fiction_service import fiction_service
|
||
from services.studio_step_respond import resolve_api_config
|
||
from utils.llm_client import LLMClient
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_llm_client = LLMClient()
|
||
_JSON_FENCE = re.compile(r"```(?:json)?\s*([\s\S]*?)```", re.IGNORECASE)
|
||
|
||
|
||
def _extract_json(raw: str) -> Dict[str, Any]:
|
||
text = (raw or "").strip()
|
||
if not text:
|
||
raise ValueError("模型返回为空")
|
||
fence = _JSON_FENCE.search(text)
|
||
if fence:
|
||
text = fence.group(1).strip()
|
||
try:
|
||
return json.loads(text)
|
||
except json.JSONDecodeError:
|
||
start = text.find("{")
|
||
end = text.rfind("}")
|
||
if start >= 0 and end > start:
|
||
return json.loads(text[start : end + 1])
|
||
raise ValueError("无法解析模型返回的 JSON")
|
||
|
||
|
||
def _validate_api_config(api_config: Dict[str, str]) -> None:
|
||
if not api_config.get("api_key"):
|
||
raise ValueError("API Key 未配置，请先在 API 配置页面保存密钥")
|
||
if not api_config.get("api_url"):
|
||
raise ValueError("API 地址未配置，请先在 API 配置页面保存 mainLLM")
|
||
|
||
|
||
def _format_catalog_for_prompt(flows: List[EmotionFlow]) -> str:
|
||
lines: List[str] = []
|
||
for flow in flows:
|
||
tags = "、".join(flow.tags or [])
|
||
lines.append(f"- id: {flow.id}\n intro: {flow.intro}\n tags: {tags}")
|
||
return "\n".join(lines) if lines else "（无可用情绪流）"
|
||
|
||
|
||
def _format_guide_global_for_prompt() -> str:
|
||
entries = fiction_service.get_guide_global_entries().entries
|
||
lines: List[str] = []
|
||
for entry in entries:
|
||
lines.append(f"[{entry.layer}] {entry.title}\n{entry.content}")
|
||
return "\n\n".join(lines) if lines else "（无全局指南）"
|
||
|
||
|
||
def _build_open_book_messages(inspiration: str) -> List[Any]:
|
||
default_settings = fiction_service.get_default_settings()
|
||
system_prompt = resolve_prompt("openBook", default_settings.prompts.openBook)
|
||
|
||
catalog = fiction_service.get_emotion_catalog()
|
||
catalog_text = _format_catalog_for_prompt(catalog.flows)
|
||
guide_global_text = _format_guide_global_for_prompt()
|
||
|
||
user_content = f"""## 全局创作指南（L0–L3）
|
||
{guide_global_text}
|
||
|
||
## 可选情绪流 catalog
|
||
{catalog_text}
|
||
|
||
## 用户创作灵感
|
||
{inspiration.strip()}
|
||
|
||
请根据以上信息优化开书方案。"""
|
||
|
||
return [
|
||
SystemMessage(content=system_prompt),
|
||
HumanMessage(content=user_content),
|
||
]
|
||
|
||
|
||
def _normalize_flow_ids(raw_ids: Any, valid_ids: set[str]) -> List[str]:
|
||
if not isinstance(raw_ids, list):
|
||
return []
|
||
result: List[str] = []
|
||
for item in raw_ids:
|
||
fid = str(item).strip()
|
||
if fid in valid_ids and fid not in result:
|
||
result.append(fid)
|
||
return result
|
||
|
||
|
||
def _parse_open_book_response(
|
||
data: Dict[str, Any], valid_flow_ids: set[str]
|
||
) -> OpenBookResult:
|
||
guide_raw = data.get("guide") or {}
|
||
guide = FictionGuideWorldbook(
|
||
persona=str(guide_raw.get("persona") or "").strip(),
|
||
highlight=str(guide_raw.get("highlight") or "").strip(),
|
||
experience=str(guide_raw.get("experience") or "").strip(),
|
||
forbiddenZones=str(guide_raw.get("forbiddenZones") or "").strip(),
|
||
)
|
||
allowed = _normalize_flow_ids(data.get("allowedFlowIds"), valid_flow_ids)
|
||
if not allowed and valid_flow_ids:
|
||
allowed = [next(iter(valid_flow_ids))]
|
||
|
||
title = str(data.get("title") or "未命名作品").strip() or "未命名作品"
|
||
optimized_intro = str(data.get("optimizedIntro") or "").strip()
|
||
|
||
return OpenBookResult(
|
||
title=title,
|
||
optimizedIntro=optimized_intro,
|
||
guide=guide,
|
||
allowedFlowIds=allowed,
|
||
)
|
||
|
||
|
||
async def run_open_book(
|
||
inspiration: str,
|
||
*,
|
||
profile_id: Optional[str] = None,
|
||
api_config: Optional[Dict[str, str]] = None,
|
||
) -> OpenBookResult:
|
||
inspiration = (inspiration or "").strip()
|
||
if not inspiration:
|
||
raise ValueError("创作灵感不能为空")
|
||
|
||
resolved = resolve_api_config(profile_id, api_config)
|
||
_validate_api_config(resolved)
|
||
|
||
catalog = fiction_service.get_emotion_catalog()
|
||
valid_flow_ids = {f.id for f in catalog.flows}
|
||
|
||
messages = _build_open_book_messages(inspiration)
|
||
response = await _llm_client.chat_completion(
|
||
messages=messages,
|
||
api_url=resolved["api_url"],
|
||
api_key=resolved["api_key"],
|
||
model=resolved.get("model", "gpt-4o-mini"),
|
||
temperature=0.7,
|
||
max_tokens=8000,
|
||
request_timeout=120,
|
||
stream=False,
|
||
)
|
||
content = response["choices"][0]["message"]["content"]
|
||
data = _extract_json(content)
|
||
return _parse_open_book_response(data, valid_flow_ids)
|