Replace temperature with reasoning effort and retry in-stream errors
- Providers get a default reasoning effort (low..max, empty = not sent); chat can override it per message. OpenAI sends reasoning_effort, Claude sends output_config.effort with adaptive thinking, and thinking blocks are replayed during the tool loop. - Show a thinking indicator while waiting for model output. - Surface in-stream upstream errors (e.g. overloaded) instead of breaking the SSE stream, and retry them up to 3 times when nothing has been emitted yet.
This commit is contained in:
eezd committed
1 parent
5db3ead883
commit
b549f3f441
9 files changed
+190
-51
No files matched your search
@@ -2,14 +2,16 @@
|
||||
|
||||
消息使用与协议无关的格式（与 ChatMessage.content 一致，外加 role）：
|
||||
{"role": "user", "text"}
|
||||
{"role": "assistant", "text", "tool_calls": [{"id", "name", "arguments"}]}
|
||||
{"role": "assistant", "text", "tool_calls": [{"id", "name", "arguments"}], "thinking_blocks"?: [...]}
|
||||
{"role": "tool", "tool_call_id", "name", "result"}
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import AsyncIterator
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
from typing import Any, Literal
|
||||
|
||||
import anthropic
|
||||
import openai
|
||||
@@ -17,9 +19,17 @@ import openai
|
||||
from alphapulse.core.crypto import decrypt
|
||||
from alphapulse.models import AiProvider
|
||||
|
||||
TIMEOUT_SECONDS = 120
|
||||
# 高思考强度时 OpenAI 格式在推理期间不推送任何内容，读超时要留足
|
||||
TIMEOUT_SECONDS = 600
|
||||
# 思考强度：两种协议共有的取值。OpenAI 另有 none/minimal，但实测经 Sub2API 会卡住后 502，不提供
|
||||
Effort = Literal["low", "medium", "high", "xhigh", "max"]
|
||||
# SDK 自带重试只针对连接错误、429 和 5xx
|
||||
MAX_RETRIES = 2
|
||||
# 流内错误（上游过载等）SDK 不重试，由这里重试；只在还没有输出内容时重试
|
||||
STREAM_RETRIES = 3
|
||||
STREAM_RETRY_DELAY_SECONDS = 2
|
||||
|
||||
logger = logging.getLogger("alphapulse.ai.llm")
|
||||
|
||||
|
||||
class LlmError(Exception):
|
||||
@@ -39,6 +49,8 @@ class Completion:
|
||||
text: str = ""
|
||||
tool_calls: list[dict] = field(default_factory=list)
|
||||
usage: dict = field(default_factory=dict)
|
||||
# Anthropic 的 thinking / redacted_thinking 块：工具调用循环中必须原样回传，否则接口报错
|
||||
thinking_blocks: list[dict] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -46,27 +58,60 @@ class TextDelta:
|
||||
text: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ThinkingStarted:
|
||||
"""模型开始思考（只有 Claude 格式会推送）；前端据此显示「思考中」"""
|
||||
|
||||
|
||||
def api_root(base_url: str) -> str:
|
||||
"""用户可能填 https://host 或 https://host/v1，统一成不带 /v1 的根地址"""
|
||||
root = base_url.rstrip("/")
|
||||
return root.removesuffix("/v1")
|
||||
|
||||
|
||||
async def stream(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec]) -> AsyncIterator[TextDelta | Completion]:
|
||||
"""先逐段产出 TextDelta，最后产出一个 Completion"""
|
||||
try:
|
||||
if provider.protocol == "openai":
|
||||
async for event in _stream_openai(provider, system, messages, tools):
|
||||
async def stream(
|
||||
provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None = None,
|
||||
) -> AsyncIterator[TextDelta | ThinkingStarted | Completion]:
|
||||
"""先逐段产出 TextDelta / ThinkingStarted，最后产出一个 Completion。effort 为空表示不传，由模型决定"""
|
||||
for attempt in range(STREAM_RETRIES + 1):
|
||||
output = False
|
||||
try:
|
||||
if provider.protocol == "openai":
|
||||
events = _stream_openai(provider, system, messages, tools, effort)
|
||||
else:
|
||||
events = _stream_anthropic(provider, system, messages, tools, effort)
|
||||
async for event in events:
|
||||
output = True
|
||||
yield event
|
||||
else:
|
||||
async for event in _stream_anthropic(provider, system, messages, tools):
|
||||
yield event
|
||||
except (openai.APIStatusError, anthropic.APIStatusError) as e:
|
||||
raise LlmError(f"模型接口返回 {e.status_code}：{e.message}") from e
|
||||
except (openai.APITimeoutError, anthropic.APITimeoutError) as e:
|
||||
raise LlmError(f"模型接口超时（{TIMEOUT_SECONDS}s）") from e
|
||||
except (openai.APIConnectionError, anthropic.APIConnectionError) as e:
|
||||
raise LlmError(f"无法连接模型接口：{e}") from e
|
||||
return
|
||||
except (openai.APIError, anthropic.APIError) as e:
|
||||
# 已经推送过内容再重试会重复输出，只重试还没有输出时的流内错误
|
||||
if output or attempt == STREAM_RETRIES or not _in_stream_error(e):
|
||||
raise _llm_error(e) from e
|
||||
logger.warning("模型流内出错，%ss 后第 %s 次重试：%s", STREAM_RETRY_DELAY_SECONDS, attempt + 1, _llm_error(e))
|
||||
await asyncio.sleep(STREAM_RETRY_DELAY_SECONDS)
|
||||
|
||||
|
||||
def _in_stream_error(e: Exception) -> bool:
|
||||
"""连接建立后上游在流内返回的错误事件（如 overloaded）。
|
||||
OpenAI SDK 抛 APIError 基类；Anthropic SDK 抛 APIStatusError，状态码是流本身的 200"""
|
||||
if isinstance(e, openai.APIError):
|
||||
return not isinstance(e, (openai.APIStatusError, openai.APIConnectionError))
|
||||
return isinstance(e, anthropic.APIStatusError) and e.status_code == 200
|
||||
|
||||
|
||||
def _llm_error(e: Exception) -> LlmError:
|
||||
if _in_stream_error(e):
|
||||
error = e.body.get("error") if isinstance(e.body, dict) else None
|
||||
message = error.get("message") if isinstance(error, dict) else None
|
||||
return LlmError(f"模型接口出错：{message or e.message}")
|
||||
if isinstance(e, (openai.APIStatusError, anthropic.APIStatusError)):
|
||||
return LlmError(f"模型接口返回 {e.status_code}：{e.message}")
|
||||
if isinstance(e, (openai.APITimeoutError, anthropic.APITimeoutError)):
|
||||
return LlmError(f"模型接口超时（{TIMEOUT_SECONDS}s）")
|
||||
if isinstance(e, (openai.APIConnectionError, anthropic.APIConnectionError)):
|
||||
return LlmError(f"无法连接模型接口：{e}")
|
||||
return LlmError(f"模型接口出错：{e}")
|
||||
|
||||
|
||||
# ---------- OpenAI ----------
|
||||
@@ -92,7 +137,7 @@ def _openai_messages(system: str, messages: list[dict]) -> list[dict]:
|
||||
return out
|
||||
|
||||
|
||||
async def _stream_openai(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec]) -> AsyncIterator[TextDelta | Completion]:
|
||||
async def _stream_openai(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None) -> AsyncIterator[TextDelta | Completion]:
|
||||
client = openai.AsyncOpenAI(
|
||||
api_key=decrypt(provider.api_key_encrypted), base_url=f"{api_root(provider.base_url)}/v1",
|
||||
timeout=TIMEOUT_SECONDS, max_retries=MAX_RETRIES,
|
||||
@@ -100,8 +145,8 @@ async def _stream_openai(provider: AiProvider, system: str, messages: list[dict]
|
||||
kwargs: dict[str, Any] = {}
|
||||
if tools:
|
||||
kwargs["tools"] = [{"type": "function", "function": {"name": t.name, "description": t.description, "parameters": t.parameters}} for t in tools]
|
||||
if provider.temperature is not None:
|
||||
kwargs["temperature"] = provider.temperature
|
||||
if effort:
|
||||
kwargs["reasoning_effort"] = effort
|
||||
result = Completion()
|
||||
# 按 index 拼接分片到达的工具调用
|
||||
calls: dict[int, dict] = {}
|
||||
@@ -139,7 +184,10 @@ def _anthropic_messages(messages: list[dict]) -> list[dict]:
|
||||
if m["role"] == "user":
|
||||
out.append({"role": "user", "content": m["text"]})
|
||||
elif m["role"] == "assistant":
|
||||
blocks: list[dict] = [{"type": "text", "text": m["text"]}] if m["text"] else []
|
||||
# 思考块必须排在最前，原样回传
|
||||
blocks: list[dict] = list(m.get("thinking_blocks") or [])
|
||||
if m["text"]:
|
||||
blocks.append({"type": "text", "text": m["text"]})
|
||||
blocks += [{"type": "tool_use", "id": c["id"], "name": c["name"], "input": c["arguments"]} for c in m.get("tool_calls") or []]
|
||||
# 空的 assistant 消息会被接口拒绝
|
||||
if blocks:
|
||||
@@ -154,7 +202,7 @@ def _anthropic_messages(messages: list[dict]) -> list[dict]:
|
||||
return out
|
||||
|
||||
|
||||
async def _stream_anthropic(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec]) -> AsyncIterator[TextDelta | Completion]:
|
||||
async def _stream_anthropic(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None) -> AsyncIterator[TextDelta | ThinkingStarted | Completion]:
|
||||
client = anthropic.AsyncAnthropic(
|
||||
api_key=decrypt(provider.api_key_encrypted), base_url=api_root(provider.base_url),
|
||||
timeout=TIMEOUT_SECONDS, max_retries=MAX_RETRIES,
|
||||
@@ -162,11 +210,13 @@ async def _stream_anthropic(provider: AiProvider, system: str, messages: list[di
|
||||
kwargs: dict[str, Any] = {}
|
||||
if tools:
|
||||
kwargs["tools"] = [{"name": t.name, "description": t.description, "input_schema": t.parameters} for t in tools]
|
||||
if provider.temperature is not None:
|
||||
# 当前 SDK 不再提供 temperature 参数，按原始字段透传
|
||||
kwargs["extra_body"] = {"temperature": provider.temperature}
|
||||
if effort:
|
||||
# 思考强度通过 output_config.effort 控制，配合自适应思考（由模型决定是否、思考多久）
|
||||
kwargs["output_config"] = {"effort": effort}
|
||||
kwargs["thinking"] = {"type": "adaptive"}
|
||||
result = Completion()
|
||||
blocks: dict[int, dict] = {}
|
||||
thinking: dict[int, dict] = {}
|
||||
async with client:
|
||||
events = await client.messages.create(
|
||||
model=provider.model, system=system, messages=_anthropic_messages(messages), max_tokens=provider.max_tokens,
|
||||
@@ -177,15 +227,26 @@ async def _stream_anthropic(provider: AiProvider, system: str, messages: list[di
|
||||
result.usage["input_tokens"] = event.message.usage.input_tokens
|
||||
elif event.type == "content_block_start" and event.content_block.type == "tool_use":
|
||||
blocks[event.index] = {"id": event.content_block.id, "name": event.content_block.name, "arguments": ""}
|
||||
elif event.type == "content_block_start" and event.content_block.type == "thinking":
|
||||
thinking[event.index] = {"type": "thinking", "thinking": event.content_block.thinking, "signature": event.content_block.signature}
|
||||
yield ThinkingStarted()
|
||||
elif event.type == "content_block_start" and event.content_block.type == "redacted_thinking":
|
||||
thinking[event.index] = {"type": "redacted_thinking", "data": event.content_block.data}
|
||||
yield ThinkingStarted()
|
||||
elif event.type == "content_block_delta":
|
||||
if event.delta.type == "text_delta":
|
||||
result.text += event.delta.text
|
||||
yield TextDelta(event.delta.text)
|
||||
elif event.delta.type == "input_json_delta":
|
||||
blocks[event.index]["arguments"] += event.delta.partial_json
|
||||
elif event.delta.type == "thinking_delta":
|
||||
thinking[event.index]["thinking"] += event.delta.thinking
|
||||
elif event.delta.type == "signature_delta":
|
||||
thinking[event.index]["signature"] += event.delta.signature
|
||||
elif event.type == "message_delta":
|
||||
result.usage["output_tokens"] = event.usage.output_tokens
|
||||
result.tool_calls = [_parse_call(b["id"], b["name"], b["arguments"]) for _, b in sorted(blocks.items())]
|
||||
result.thinking_blocks = [b for _, b in sorted(thinking.items())]
|
||||
yield result
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user