src/llm/client.py
Rama main · réplica de lectura del repositorio original.
"""LLM 客户端 —— 通过 aiohttp 直连 OpenAI 兼容 API。"""
from __future__ import annotations
import asyncio
import json
import logging
import re
from dataclasses import dataclass, field
from typing import Any
import aiohttp
from src.llm.openai_transport import OpenAIChatTransport
from src.llm.generation_options import apply_anthropic_options, validate_generation_options
logger = logging.getLogger("trpg")
MAX_RETRIES = 3 # 总共尝试次数（含首次）
BASE_DELAY = 2.0 # 基础重试间隔（秒）
LENGTH_RETRY_FACTOR = 2 # finish_reason=length 时重试放大 max_tokens 的倍数
LENGTH_RETRY_MAX_MULT = 4 # 最大放大到原始 max_tokens 的多少倍
_THINK_BLOCK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
def strip_reasoning_from_content(text: str) -> str:
"""删除 OpenAI 兼容代理混入正文的 reasoning 思考块，返回最终正文。
处理完整 ``<think>…</think>``（含多个块）、孤立 ``</think>`` 以及未闭合的
``<think>`` 尾部；正常正文原样保留。清洗后没有正文时抛 ValueError，由
call()/call_stream() 的既有失败重试逻辑接手，确保思考内容不会作为正文
保存进 GM log 或展示给玩家。
"""
cleaned = _THINK_BLOCK_RE.sub("", str(text or ""))
cleaned = cleaned.replace("</think>", "")
cleaned = cleaned.split("<think>", 1)[0].strip()
if not cleaned:
raise ValueError("模型未返回最终正文")
return cleaned
def _disable_thinking_for_deepseek_v4(provider: "ProviderConfig") -> bool:
"""DiceFrame 对官方 DeepSeek V4 使用非思考模式。
DeepSeek V4 默认开启思考模式；当前客户端不消费 ``reasoning_content``，
该模式还不接受强制 ``tool_choice``，会让普通正文和严格 JSON 的输出预算
被推理耗尽。仅针对官方端点和 V4 模型显式关闭，避免向其他 OpenAI 兼容
供应商发送私有参数。
"""
return (
provider.model_name.strip().lower().startswith("deepseek-v4")
and "api.deepseek.com" in provider.base_url.strip().lower()
)
def length_retry_budgets(base_max_tokens: int) -> tuple[int, ...]:
"""返回输出截断时共用的 token 预算序列。"""
if base_max_tokens <= 0:
raise ValueError("base_max_tokens 必须大于 0")
budgets = [base_max_tokens]
limit = base_max_tokens * LENGTH_RETRY_MAX_MULT
while len(budgets) < MAX_RETRIES:
bumped = min(budgets[-1] * LENGTH_RETRY_FACTOR, limit)
if bumped <= budgets[-1]:
break
budgets.append(bumped)
return tuple(budgets)
# 按 HTTP 状态码的退避策略
_RETRY_BACKOFF: dict[int, float] = {
429: 5.0, # Rate Limit → 较长等待
503: 3.0, # Service Unavailable → 中等等待
502: 2.0, # Bad Gateway
504: 2.0, # Gateway Timeout
}
_RETRYABLE_STATUSES = frozenset(_RETRY_BACKOFF) | {408} # 408 Timeout 也可重试
class OutputTruncatedError(ValueError):
"""模型输出被 max_tokens 截断（finish_reason=length / stop_reason=max_tokens）。
即使已经返回部分正文也不能当作完整结果；call() 据此在重试时自动提高
max_tokens，而不是保存缺少结尾或结构化状态的半截回复。
"""
def __init__(self, finish_reason: str = "length"):
self.finish_reason = finish_reason
super().__init__(f"模型未返回最终正文 (finish_reason={finish_reason})")
@dataclass
class ProviderConfig:
"""模型供应配置。"""
provider_name: str
base_url: str
api_key: str
model_name: str
api_format: str = "openai"
fallback: bool = False
@dataclass
class LLMResponse:
"""LLM 调用结果。"""
content: str # 完整响应文本
narration: str # 叙事文本（JSON 块之前的部分）
state_update: dict | None # 解析出的状态更新
memory_delta: dict | None # 解析出的记忆变更
info_asymmetry: dict | None # 解析出的信息不对称
plot_update: dict | None # 解析出的剧情推进
total_tokens: int # 实际消耗的 token 数
is_narration_only: bool # JSON 解析失败，仅叙事
provider_used: str # 实际使用的供应商名称
token_budget_initial: int = 0 # 本次调用最初配置的最大输出 token
token_budget_used: int = 0 # 成功响应实际使用的最大输出 token 档位
system_notices: list[str] = field(default_factory=list) # 不属于 GM 正文的系统提示
story: dict[str, Any] | None = None # Validated public presentation, independent of rule effects.
@dataclass
class LLMToolResponse:
"""一次工具规划调用的结构化结果。"""
tool_calls: list[dict[str, Any]]
total_tokens: int
provider_used: str
native_tools: bool = True
content: str = ""
class LLMToolCallError(RuntimeError):
"""Provider failure with HTTP facts preserved for application feedback."""
def __init__(self, message: str, status_codes: frozenset[int]):
super().__init__(message)
self.status_codes = status_codes
class LLMClient:
"""OpenAI 兼容 API 的异步 HTTP 客户端。
支持多供应商配置、自动重试、失败降级到备用模型。
"""
def __init__(
self,
providers: list[ProviderConfig],
default: str,
proxy_url: str = "",
request_timeout_seconds: float = 120,
generation_options: dict[str, Any] | None = None,
):
if not providers:
raise ValueError("至少需要配置一个模型供应商")
if not 10 <= float(request_timeout_seconds) <= 600:
raise ValueError("模型请求超时必须在 10–600 秒之间")
self.providers = {p.provider_name: p for p in providers}
self.default = default if default in self.providers else providers[0].provider_name
self.proxy_url = proxy_url.strip()
self.request_timeout_seconds = float(request_timeout_seconds)
self.request_timeout = aiohttp.ClientTimeout(total=self.request_timeout_seconds)
self._session: aiohttp.ClientSession | None = None
self._native_tool_unsupported: set[str] = set()
# 强制 tool_choice 被拒但 tool_choice=auto 可用的供应商（如思考模式模型）
self._native_tool_auto: set[str] = set()
self.generation_options = validate_generation_options({} if generation_options is None else generation_options)
self._openai_transport = OpenAIChatTransport(self.generation_options)
def _anthropic_body(self, body: dict[str, Any]) -> dict[str, Any]:
apply_anthropic_options(body, self.generation_options)
return body
async def _get_session(self) -> aiohttp.ClientSession:
"""获取或懒创建复用的 HTTP session。"""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession()
return self._session
async def close(self) -> None:
if self._session and not self._session.closed:
await self._session.close()
self._session = None
async def call_tools(
self,
system_prompt: str,
user_message: str,
*,
tools: list[dict[str, Any]],
max_tokens: int = 4096,
temperature: float = 0.1,
force_tool_choice: bool = True,
) -> LLMToolResponse:
"""调用模型工具；不支持原生 tools 的供应商自动回退严格 JSON。
回退仍只产生工具参数，绝不让模型产生或指定骰面。所有调用失败时抛出，
由上层决定是否进入离线兼容路径。
思考模式模型常拒绝强制 tool_choice 但接受 tool_choice=auto：此类供应商
先用 auto 重试原生调用，仍失败才走 JSON 回退；JSON 回退截断时按
length_retry_budgets 放大 max_tokens 重试。显式 ``force_tool_choice=False``
时始终请求原生 auto；该选择本身不会把供应商记为不支持强制调用。
"""
primary = self.providers[self.default]
ordered = [primary] + [
provider for provider in self.providers.values()
if provider.fallback and provider.provider_name != primary.provider_name
]
errors: list[str] = []
status_codes: set[int] = set()
for provider in ordered:
forced = force_tool_choice and provider.provider_name not in self._native_tool_unsupported
if not force_tool_choice or forced or provider.provider_name in self._native_tool_auto:
try:
native_result = await self._call_tools_one(
provider,
system_prompt,
user_message,
tools,
temperature,
max_tokens,
force_tool_choice=forced,
)
if force_tool_choice and not native_result.tool_calls:
raise ValueError("供应商未返回必需的工具调用")
return native_result
except Exception as exc:
if isinstance(exc, aiohttp.ClientResponseError):
status_codes.add(exc.status)
errors.append(f"{provider.provider_name}: {exc}")
if forced and "does not support this tool_choice" in str(exc):
self._native_tool_unsupported.add(provider.provider_name)
try:
result = await self._call_tools_one(
provider,
system_prompt,
user_message,
tools,
temperature,
max_tokens,
force_tool_choice=False,
)
if not result.tool_calls:
raise ValueError("供应商未返回必需的工具调用")
self._native_tool_auto.add(provider.provider_name)
return result
except Exception as auto_exc:
if isinstance(auto_exc, aiohttp.ClientResponseError):
status_codes.add(auto_exc.status)
errors.append(f"{provider.provider_name} (tool_choice=auto): {auto_exc}")
logger.warning("模型原生工具调用失败，尝试 JSON 回退 (%s): %s", provider.provider_name, exc)
try:
tool_names = [
str(tool.get("function", {}).get("name") or "tool")
for tool in tools
]
if force_tool_choice:
tool_instruction = (
f"name={tool_names[0]} and arguments matching that tool's schema."
)
else:
tool_instruction = (
"name matching one of the available tools "
f"({', '.join(tool_names)}) and arguments matching that tool's schema."
)
fallback_prompt = (
system_prompt
+ "\n\nNative function calling is unavailable. Adjudicate the supplied actions first, "
+ "then return only one JSON object with a tool_calls array. Each item must contain "
+ tool_instruction + " "
+ "Never copy an empty structural example when the supplied context supports a concrete result. "
+ "Required tool schema: "
+ json.dumps(tools, ensure_ascii=False, separators=(",", ":"))
)
budgets = length_retry_budgets(max_tokens)
response = None
for attempt, budget in enumerate(budgets, 1):
try:
response = await self._call_one(
provider,
fallback_prompt,
user_message,
temperature,
budget,
True,
)
break
except OutputTruncatedError:
if attempt >= len(budgets):
raise
logger.warning(
"JSON 回退输出截断，提高 max_tokens 重试 (%s): %d -> %d",
provider.provider_name, budget, budgets[attempt],
)
await asyncio.sleep(BASE_DELAY * 0.5)
calls = _parse_json_tool_calls(response.content, allow_legacy=force_tool_choice)
if not force_tool_choice:
calls = _validate_tool_calls(calls, tools)
if not calls:
raise ValueError("模型未返回必需的工具调用")
return LLMToolResponse(
tool_calls=calls,
total_tokens=response.total_tokens,
provider_used=provider.provider_name,
native_tools=False,
)
except Exception as exc:
if isinstance(exc, aiohttp.ClientResponseError):
status_codes.add(exc.status)
errors.append(f"{provider.provider_name} JSON fallback: {exc}")
logger.warning("模型 JSON 工具回退失败 (%s): %s", provider.provider_name, exc)
raise LLMToolCallError("所有模型工具调用均失败: " + "; ".join(errors), frozenset(status_codes))
async def _call_tools_one(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
tools: list[dict[str, Any]],
temperature: float,
max_tokens: int,
force_tool_choice: bool = True,
) -> LLMToolResponse:
if (provider.api_format or "openai").strip().lower() == "anthropic":
return await self._call_tools_anthropic(
provider, system_prompt, user_message, tools, temperature, max_tokens,
force_tool_choice=force_tool_choice,
)
return await self._call_tools_openai(
provider, system_prompt, user_message, tools, temperature, max_tokens,
force_tool_choice=force_tool_choice,
)
async def _call_tools_openai(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
tools: list[dict[str, Any]],
temperature: float,
max_tokens: int,
force_tool_choice: bool = True,
) -> LLMToolResponse:
url = provider.base_url.rstrip("/")
if not url.endswith("/chat/completions"):
url += "/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {provider.api_key}",
}
body: dict[str, Any] = {
"model": provider.model_name,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
"temperature": temperature,
"max_tokens": max_tokens,
"tools": tools,
}
if _disable_thinking_for_deepseek_v4(provider):
body["thinking"] = {"type": "disabled"}
if force_tool_choice:
body["tool_choice"] = {
"type": "function",
"function": {"name": str(tools[0].get("function", {}).get("name") or "")},
}
else:
body["tool_choice"] = "auto"
body["parallel_tool_calls"] = False
session = await self._get_session()
request_kwargs = {"proxy": self.proxy_url} if self.proxy_url else {}
async with self._openai_transport.post(
session, url,
json=body,
headers=headers,
timeout=self.request_timeout,
**request_kwargs,
) as resp:
if resp.status != 200:
error_text = await resp.text()
raise aiohttp.ClientResponseError(
resp.request_info,
resp.history,
status=resp.status,
message=error_text[:300],
headers=resp.headers,
)
data = await resp.json()
message = data.get("choices", [{}])[0].get("message", {})
calls = _normalize_openai_tool_calls(message.get("tool_calls") or [])
content = _openai_text_content(message.get("content"))
if not calls and content:
if not force_tool_choice:
explicit_calls = _parse_explicit_json_tool_calls(content)
if explicit_calls is not None:
return LLMToolResponse(
tool_calls=explicit_calls,
total_tokens=int(data.get("usage", {}).get("total_tokens", 0) or 0),
provider_used=provider.provider_name,
native_tools=True,
content=content,
)
return LLMToolResponse(
tool_calls=[],
total_tokens=int(data.get("usage", {}).get("total_tokens", 0) or 0),
provider_used=provider.provider_name,
native_tools=True,
content=content,
)
calls = _parse_json_tool_calls(content)
if not calls:
raise ValueError("供应商未返回强制工具调用")
return LLMToolResponse(
tool_calls=calls,
total_tokens=int(data.get("usage", {}).get("total_tokens", 0) or 0),
provider_used=provider.provider_name,
native_tools=True,
content=content,
)
async def _call_tools_anthropic(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
tools: list[dict[str, Any]],
temperature: float,
max_tokens: int,
force_tool_choice: bool = True,
) -> LLMToolResponse:
headers = {
"Content-Type": "application/json",
"x-api-key": provider.api_key,
"anthropic-version": "2023-06-01",
}
anthropic_tools = []
for tool in tools:
function = tool.get("function", {})
anthropic_tools.append({
"name": function.get("name", ""),
"description": function.get("description", ""),
"input_schema": function.get("parameters", {"type": "object"}),
})
body = {
"model": provider.model_name,
"system": system_prompt,
"messages": [{"role": "user", "content": user_message}],
"temperature": temperature,
"max_tokens": max_tokens,
"tools": anthropic_tools,
}
if force_tool_choice:
body["tool_choice"] = {
"type": "tool",
"name": str(anthropic_tools[0].get("name") or ""),
}
else:
body["tool_choice"] = {"type": "auto"}
session = await self._get_session()
request_kwargs = {"proxy": self.proxy_url} if self.proxy_url else {}
async with session.post(
_anthropic_messages_url(provider.base_url),
json=self._anthropic_body(body),
headers=headers,
timeout=self.request_timeout,
**request_kwargs,
) as resp:
if resp.status != 200:
error_text = await resp.text()
raise aiohttp.ClientResponseError(
resp.request_info,
resp.history,
status=resp.status,
message=error_text[:300],
headers=resp.headers,
)
data = await resp.json()
calls = [
{"name": str(block.get("name") or ""), "arguments": block.get("input") or {}}
for block in data.get("content", [])
if isinstance(block, dict) and block.get("type") == "tool_use"
]
content = _anthropic_text_content(data)
if not calls and content and not force_tool_choice:
return LLMToolResponse(
tool_calls=[],
total_tokens=(
int(data.get("usage", {}).get("input_tokens", 0) or 0)
+ int(data.get("usage", {}).get("output_tokens", 0) or 0)
),
provider_used=provider.provider_name,
native_tools=True,
content=content,
)
if not calls:
raise ValueError("供应商未返回强制工具调用")
usage = data.get("usage", {})
total_tokens = int(usage.get("input_tokens", 0) or 0) + int(usage.get("output_tokens", 0) or 0)
return LLMToolResponse(calls, total_tokens, provider.provider_name, True, content)
async def call(
self,
system_prompt: str,
user_message: str,
*,
temperature: float = 0.7,
max_tokens: int = 4096,
force_provider: str | None = None,
json_mode: bool = False,
) -> LLMResponse:
"""调用 LLM，失败时自动降级到 fallback 模型。
Args:
system_prompt: GM 系统提示词
user_message: 拼接好的上下文
temperature: 生成温度
max_tokens: 最大输出 token 数
force_provider: 强制使用指定供应商（跳过低负载均衡）
json_mode: 启用 JSON 模式（DeepSeek/OpenAI 兼容的 structured output）
Returns:
LLMResponse
Raises:
RuntimeError: 所有供应商均失败
"""
primary = self.providers[force_provider or self.default]
ordered = [primary] + [
p for p in self.providers.values()
if p.fallback and p.provider_name != primary.provider_name
]
last_error = None
last_backoff = BASE_DELAY
retry_budgets = length_retry_budgets(max_tokens)
current_max_tokens = retry_budgets[0]
for attempt_num in range(1, MAX_RETRIES + 1):
for provider in ordered:
try:
response = await self._call_one(
provider, system_prompt, user_message,
temperature, current_max_tokens, json_mode,
)
response.token_budget_initial = max_tokens
response.token_budget_used = current_max_tokens
return response
except OutputTruncatedError as exc:
last_error = exc
logger.warning(
"LLM 输出被 max_tokens 截断 (attempt=%d, provider=%s, max_tokens=%d): %s",
attempt_num, provider.provider_name, current_max_tokens, exc,
)
continue
except aiohttp.ClientResponseError as exc:
last_error = exc
status = exc.status
if status in _RETRYABLE_STATUSES:
last_backoff = _RETRY_BACKOFF.get(status, BASE_DELAY) * attempt_num
logger.warning(
"LLM HTTP %d (attempt=%d, provider=%s): %s, %0.1fs后重试",
status, attempt_num, provider.provider_name, exc, last_backoff,
)
else:
logger.warning(
"LLM HTTP %d (attempt=%d, provider=%s): %s (不可重试，跳过该供应商)",
status, attempt_num, provider.provider_name, exc,
)
continue
except asyncio.TimeoutError as exc:
last_error = exc
last_backoff = BASE_DELAY * attempt_num * 0.5
logger.warning(
"LLM 超时 (attempt=%d, provider=%s): %0.1fs后重试",
attempt_num, provider.provider_name, last_backoff,
)
except Exception as exc:
last_error = exc
logger.warning(
"LLM 调用失败 (attempt=%d, provider=%s): %s",
attempt_num, provider.provider_name, exc,
)
continue
if attempt_num < MAX_RETRIES:
if isinstance(last_error, OutputTruncatedError):
bumped = retry_budgets[
min(attempt_num, len(retry_budgets) - 1)
]
if bumped > current_max_tokens:
logger.info(
"输出截断，下次重试提高 max_tokens: %d -> %d",
current_max_tokens, bumped,
)
current_max_tokens = bumped
await asyncio.sleep(BASE_DELAY * 0.5)
else:
await asyncio.sleep(last_backoff)
raise RuntimeError(f"所有模型供应商均调用失败: {last_error}") from last_error
async def call_stream(
self,
system_prompt: str,
user_message: str,
*,
temperature: float = 0.7,
max_tokens: int = 4096,
force_provider: str | None = None,
json_mode: bool = False,
on_delta=None,
) -> LLMResponse:
"""流式调用 LLM，逐段文本通过 on_delta 回调推送，返回与 call() 相同的 LLMResponse。
on_delta 为可选的 async callable，签名为 ``async def on_delta(text: str) -> None``；
每收到一段正文就调用一次，供上层做实时打字机展示。返回值与 call() 一致，
内部累积完整正文后走同一个 _to_response 解析，保证流式/非流式结果同构。
供应商在开始流式前失败（连接/HTTP 错误/超时）时按 fallback 顺序与退避重试，
行为与 call() 一致；流式过程中途失败则抛出，由调用方决定是否重试。
finish_reason=length 时即使已有部分正文也抛 OutputTruncatedError，不在内部
放大预算，由调用方提高 max_tokens 后重新调用（与 call() 的截断处理对齐）。
"""
primary = self.providers[force_provider or self.default]
ordered = [primary] + [
p for p in self.providers.values()
if p.fallback and p.provider_name != primary.provider_name
]
last_error = None
last_backoff = BASE_DELAY
for attempt_num in range(1, MAX_RETRIES + 1):
for provider in ordered:
try:
content, total_tokens, _finish_reason = await self._stream_one(
provider, system_prompt, user_message,
temperature, max_tokens, json_mode, on_delta,
)
response = self._to_response(content, total_tokens, provider.provider_name)
response.token_budget_initial = max_tokens
response.token_budget_used = max_tokens
return response
except OutputTruncatedError:
raise
except aiohttp.ClientResponseError as exc:
last_error = exc
status = exc.status
if status in _RETRYABLE_STATUSES:
last_backoff = _RETRY_BACKOFF.get(status, BASE_DELAY) * attempt_num
logger.warning(
"LLM 流式 HTTP %d (attempt=%d, provider=%s): %s, %0.1fs后重试",
status, attempt_num, provider.provider_name, exc, last_backoff,
)
else:
logger.warning(
"LLM 流式 HTTP %d (attempt=%d, provider=%s): %s (不可重试，跳过该供应商)",
status, attempt_num, provider.provider_name, exc,
)
continue
except asyncio.TimeoutError as exc:
last_error = exc
last_backoff = BASE_DELAY * attempt_num * 0.5
logger.warning(
"LLM 流式超时 (attempt=%d, provider=%s): %0.1fs后重试",
attempt_num, provider.provider_name, last_backoff,
)
except Exception as exc:
last_error = exc
logger.warning(
"LLM 流式调用失败 (attempt=%d, provider=%s): %s",
attempt_num, provider.provider_name, exc,
)
continue
if attempt_num < MAX_RETRIES:
await asyncio.sleep(last_backoff)
raise RuntimeError(f"所有模型供应商均流式调用失败: {last_error}") from last_error
async def _call_one(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
temperature: float,
max_tokens: int,
json_mode: bool = False,
) -> LLMResponse:
api_format = (provider.api_format or "openai").strip().lower()
if api_format == "anthropic":
return await self._call_anthropic(
provider, system_prompt, user_message,
temperature, max_tokens, json_mode,
)
return await self._call_openai_compatible(
provider, system_prompt, user_message,
temperature, max_tokens, json_mode,
)
async def _call_openai_compatible(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
temperature: float,
max_tokens: int,
json_mode: bool = False,
) -> LLMResponse:
url = provider.base_url.rstrip("/")
if not url.endswith("/chat/completions"):
url += "/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {provider.api_key}",
}
body = {
"model": provider.model_name,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
"temperature": temperature,
"max_tokens": max_tokens,
}
if _disable_thinking_for_deepseek_v4(provider):
body["thinking"] = {"type": "disabled"}
# JSON 模式：DeepSeek/OpenAI 兼容的 structured output
if json_mode:
body["response_format"] = {"type": "json_object"}
session = await self._get_session()
request_kwargs = {"proxy": self.proxy_url} if self.proxy_url else {}
async with self._openai_transport.post(session, url, json=body, headers=headers,
timeout=self.request_timeout,
**request_kwargs) as resp:
if resp.status != 200:
error_text = await resp.text()
raise aiohttp.ClientResponseError(
resp.request_info, resp.history,
status=resp.status, message=error_text[:300],
headers=resp.headers,
)
data = await resp.json()
choice = data["choices"][0]
finish_reason = str(choice.get("finish_reason") or "unknown")
if finish_reason == "length":
raise OutputTruncatedError(finish_reason)
# 部分代理会把 <think>…</think> 思考混进 content，先清洗再判空
content = strip_reasoning_from_content(choice["message"].get("content") or "")
total_tokens = data.get("usage", {}).get("total_tokens", 0)
return self._to_response(content, total_tokens, provider.provider_name)
async def _call_anthropic(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
temperature: float,
max_tokens: int,
json_mode: bool = False,
) -> LLMResponse:
url = _anthropic_messages_url(provider.base_url)
system = system_prompt
if json_mode:
system = f"{system_prompt}\n\nReturn only valid JSON. Do not wrap it in Markdown."
headers = {
"Content-Type": "application/json",
"x-api-key": provider.api_key,
"anthropic-version": "2023-06-01",
}
body = {
"model": provider.model_name,
"system": system,
"messages": [
{"role": "user", "content": user_message},
],
"max_tokens": max_tokens,
}
session = await self._get_session()
request_kwargs = {"proxy": self.proxy_url} if self.proxy_url else {}
async with session.post(url, json=self._anthropic_body(body), headers=headers,
timeout=self.request_timeout,
**request_kwargs) as resp:
if resp.status != 200:
error_text = await resp.text()
raise aiohttp.ClientResponseError(
resp.request_info, resp.history,
status=resp.status, message=error_text[:300],
headers=resp.headers,
)
data = await resp.json()
content = _anthropic_text_content(data)
if data.get("stop_reason") == "max_tokens":
raise OutputTruncatedError("max_tokens")
if not content.strip():
raise ValueError(
f"模型未返回最终正文 (stop_reason={data.get('stop_reason') or 'unknown'})"
)
usage = data.get("usage", {})
total_tokens = int(usage.get("input_tokens", 0) or 0) + int(usage.get("output_tokens", 0) or 0)
return self._to_response(content, total_tokens, provider.provider_name)
async def _stream_one(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
temperature: float,
max_tokens: int,
json_mode: bool,
on_delta=None,
) -> tuple[str, int, str]:
api_format = (provider.api_format or "openai").strip().lower()
if api_format == "anthropic":
return await self._stream_anthropic(
provider, system_prompt, user_message,
temperature, max_tokens, json_mode, on_delta,
)
return await self._stream_openai_compatible(
provider, system_prompt, user_message,
temperature, max_tokens, json_mode, on_delta,
)
async def _stream_openai_compatible(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
temperature: float,
max_tokens: int,
json_mode: bool = False,
on_delta=None,
) -> tuple[str, int, str]:
url = provider.base_url.rstrip("/")
if not url.endswith("/chat/completions"):
url += "/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {provider.api_key}",
}
body = {
"model": provider.model_name,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
"temperature": temperature,
"max_tokens": max_tokens,
"stream": True,
"stream_options": {"include_usage": True},
}
if _disable_thinking_for_deepseek_v4(provider):
body["thinking"] = {"type": "disabled"}
# JSON 模式：DeepSeek/OpenAI 兼容的 structured output
if json_mode:
body["response_format"] = {"type": "json_object"}
session = await self._get_session()
request_kwargs = {"proxy": self.proxy_url} if self.proxy_url else {}
content_parts: list[str] = []
finish_reason = "stop"
total_tokens = 0
async with self._openai_transport.post(session, url, json=body, headers=headers,
timeout=self.request_timeout,
**request_kwargs) as resp:
if resp.status != 200:
error_text = await resp.text()
raise aiohttp.ClientResponseError(
resp.request_info, resp.history,
status=resp.status, message=error_text[:300],
headers=resp.headers,
)
async for raw_line in resp.content:
line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n")
if not line.startswith("data:"):
continue
payload = line[len("data:"):].strip()
if not payload:
continue
if payload == "[DONE]":
break
try:
chunk = json.loads(payload)
except json.JSONDecodeError:
continue
choices = chunk.get("choices") or []
if choices:
choice = choices[0]
delta = choice.get("delta") or {}
text = delta.get("content") or ""
if text:
content_parts.append(text)
if on_delta:
await on_delta(text)
if choice.get("finish_reason"):
finish_reason = choice["finish_reason"]
usage = chunk.get("usage")
if isinstance(usage, dict):
total_tokens = int(usage.get("total_tokens", 0) or 0)
content = "".join(content_parts)
if finish_reason == "length":
raise OutputTruncatedError(finish_reason)
# on_delta 已按原始增量转发（玩家可见内容由上层 _NarrationDeltaFilter 过滤）；
# 累积正文同样清洗，保证 response.content / GM log 不含思考内容
content = strip_reasoning_from_content(content)
return content, total_tokens, finish_reason
async def _stream_anthropic(
self,
provider: ProviderConfig,
system_prompt: str,
user_message: str,
temperature: float,
max_tokens: int,
json_mode: bool = False,
on_delta=None,
) -> tuple[str, int, str]:
url = _anthropic_messages_url(provider.base_url)
system = system_prompt
if json_mode:
system = f"{system_prompt}\n\nReturn only valid JSON. Do not wrap it in Markdown."
headers = {
"Content-Type": "application/json",
"x-api-key": provider.api_key,
"anthropic-version": "2023-06-01",
}
body = {
"model": provider.model_name,
"system": system,
"messages": [
{"role": "user", "content": user_message},
],
"max_tokens": max_tokens,
"stream": True,
}
session = await self._get_session()
request_kwargs = {"proxy": self.proxy_url} if self.proxy_url else {}
content_parts: list[str] = []
stop_reason = "end_turn"
input_tokens = 0
output_tokens = 0
async with session.post(url, json=self._anthropic_body(body), headers=headers,
timeout=self.request_timeout,
**request_kwargs) as resp:
if resp.status != 200:
error_text = await resp.text()
raise aiohttp.ClientResponseError(
resp.request_info, resp.history,
status=resp.status, message=error_text[:300],
headers=resp.headers,
)
async for raw_line in resp.content:
line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n")
if not line.startswith("data:"):
continue
payload = line[len("data:"):].strip()
if not payload:
continue
try:
chunk = json.loads(payload)
except json.JSONDecodeError:
continue
etype = chunk.get("type")
if etype == "message_start":
message = chunk.get("message") or {}
usage = message.get("usage") or {}
input_tokens = int(usage.get("input_tokens", 0) or 0)
elif etype == "content_block_delta":
delta = chunk.get("delta") or {}
if delta.get("type") == "text_delta":
text = delta.get("text") or ""
if text:
content_parts.append(text)
if on_delta:
await on_delta(text)
elif etype == "message_delta":
delta = chunk.get("delta") or {}
if delta.get("stop_reason"):
stop_reason = delta["stop_reason"]
usage = chunk.get("usage") or {}
output_tokens = int(usage.get("output_tokens", 0) or 0)
content = "".join(content_parts)
if stop_reason == "max_tokens":
raise OutputTruncatedError("max_tokens")
if not content.strip():
raise ValueError(f"模型未返回最终正文 (stop_reason={stop_reason})")
total_tokens = input_tokens + output_tokens
return content, total_tokens, stop_reason
@staticmethod
def _to_response(content: str, total_tokens: int, provider_used: str) -> LLMResponse:
# 解析输出
from .parser import parse_llm_response
result = parse_llm_response(content)
return LLMResponse(
content=content,
narration=result.narration,
state_update=result.state_update,
memory_delta=result.memory_delta,
info_asymmetry=result.info_asymmetry,
plot_update=result.plot_update,
total_tokens=total_tokens,
is_narration_only=result.is_narration_only,
provider_used=provider_used,
)
def _anthropic_messages_url(base_url: str) -> str:
url = (base_url or "https://api.anthropic.com").strip().rstrip("/")
if url.endswith("/v1/messages"):
return url
if url.endswith("/v1"):
return f"{url}/messages"
return f"{url}/v1/messages"
def _anthropic_text_content(data: dict) -> str:
parts: list[str] = []
for block in data.get("content", []):
if isinstance(block, dict) and block.get("type") == "text":
parts.append(str(block.get("text") or ""))
return "\n".join(part for part in parts if part).strip()
def _openai_text_content(value: Any) -> str:
"""Extract text from string or content-part OpenAI message payloads."""
if isinstance(value, str):
return value.strip()
if not isinstance(value, list):
return ""
parts: list[str] = []
for block in value:
if isinstance(block, dict) and block.get("type") in {"text", "output_text"}:
text = str(block.get("text") or "").strip()
if text:
parts.append(text)
return "\n".join(parts).strip()
def _tool_arguments(value: Any) -> dict[str, Any]:
if isinstance(value, dict):
return value
if isinstance(value, str):
parsed = json.loads(value)
if isinstance(parsed, dict):
return parsed
raise ValueError("工具 arguments 必须是 JSON object")
def _normalize_openai_tool_calls(raw_calls: list[Any]) -> list[dict[str, Any]]:
calls: list[dict[str, Any]] = []
for raw in raw_calls:
if not isinstance(raw, dict):
continue
function = raw.get("function") if isinstance(raw.get("function"), dict) else raw
name = str(function.get("name") or "").strip()
if not name:
continue
calls.append({"name": name, "arguments": _tool_arguments(function.get("arguments") or {})})
return calls
def _parse_json_tool_calls(content: str, *, allow_legacy: bool = True) -> list[dict[str, Any]]:
text = str(content or "").strip()
if text.startswith("```"):
text = text.split("\n", 1)[-1]
text = text.rsplit("```", 1)[0].strip()
payload = json.loads(text)
raw_calls = payload.get("tool_calls") if isinstance(payload, dict) else None
if allow_legacy and raw_calls is None and isinstance(payload, dict) and "checks" in payload:
raw_calls = [{"name": "dice_checks", "arguments": payload}]
if not isinstance(raw_calls, list):
raise ValueError("JSON 回退缺少 tool_calls")
return _normalize_openai_tool_calls(raw_calls)
def _parse_explicit_json_tool_calls(content: str) -> list[dict[str, Any]] | None:
"""Parse only JSON content with an explicit top-level ``tool_calls`` field."""
text = str(content or "").strip()
if text.startswith("```"):
text = text.split("\n", 1)[-1]
text = text.rsplit("```", 1)[0].strip()
try:
payload = json.loads(text)
except (TypeError, json.JSONDecodeError):
return None
if not isinstance(payload, dict) or "tool_calls" not in payload:
return None
raw_calls = payload.get("tool_calls")
if not isinstance(raw_calls, list):
raise ValueError("JSON 工具调用的 tool_calls 必须是 array")
calls = _normalize_openai_tool_calls(raw_calls)
if not calls:
raise ValueError("JSON 工具调用缺少有效调用")
return calls
def _validate_tool_calls(
calls: list[dict[str, Any]],
tools: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""Validate fallback calls against the registered tool names and required fields."""
schemas: dict[str, dict[str, Any]] = {}
for tool in tools:
function = tool.get("function") if isinstance(tool, dict) else None
if not isinstance(function, dict):
continue
name = str(function.get("name") or "").strip()
if name:
schema = function.get("parameters")
schemas[name] = schema if isinstance(schema, dict) else {}
if not calls:
raise ValueError("JSON 回退缺少 tool_calls")
for call in calls:
name = str(call.get("name") or "").strip()
if name not in schemas:
raise ValueError(f"JSON 回退调用了未注册工具: {name or '<empty>'}")
arguments = call.get("arguments")
if not isinstance(arguments, dict):
raise ValueError(f"JSON 回退工具参数不是 object: {name}")
return calls