diff --git a/backend/alembic/versions/20261014_5c2e8f1a7b3d_provider_reasoning_effort.py b/backend/alembic/versions/20261014_5c2e8f1a7b3d_provider_reasoning_effort.py new file mode 100644 index 0000000..54d2f0e --- /dev/null +++ b/backend/alembic/versions/20261014_5c2e8f1a7b3d_provider_reasoning_effort.py @@ -0,0 +1,31 @@ +"""provider reasoning effort replaces temperature + +Revision ID: 5c2e8f1a7b3d +Revises: 8349ae014ddd +Create Date: 2026-10-14 14:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '5c2e8f1a7b3d' +down_revision: Union[str, Sequence[str], None] = '8349ae014ddd' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # 用户决定不再设置 temperature（2026-10-14），改为思考强度 + op.drop_column('ai_providers', 'temperature') + op.add_column('ai_providers', sa.Column('reasoning_effort', sa.String(length=8), nullable=True)) + + +def downgrade() -> None: + """Downgrade schema.""" + op.drop_column('ai_providers', 'reasoning_effort') + op.add_column('ai_providers', sa.Column('temperature', sa.Float(), nullable=True)) diff --git a/backend/src/alphapulse/ai/llm.py b/backend/src/alphapulse/ai/llm.py index ee3dfbe..4c67308 100644 --- a/backend/src/alphapulse/ai/llm.py +++ b/backend/src/alphapulse/ai/llm.py @@ -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 diff --git a/backend/src/alphapulse/api/chat.py b/backend/src/alphapulse/api/chat.py index da9dcf5..d9ae6e6 100644 --- a/backend/src/alphapulse/api/chat.py +++ b/backend/src/alphapulse/api/chat.py @@ -53,6 +53,8 @@ class SessionOut(BaseModel): class SendIn(BaseModel): text: Annotated[str, Field(min_length=1, max_length=20000)] provider_id: int | None = None + # 本轮思考强度；为空时用模型配置里的默认值 + reasoning_effort: llm.Effort | None = None class ResetIn(BaseModel): @@ -121,7 +123,7 @@ async def get_chat_session(session: Annotated[AsyncSession, Depends(get_session) @router.post("/messages", response_class=EventSourceResponse) async def send_message(body: SendIn) -> AsyncIterator[ServerSentEvent]: - """事件：prompt、delta、tool_call、tool_result、skill_draft、context_updated、usage、error、done""" + """事件：prompt、thinking、delta、tool_call、tool_result、skill_draft、context_updated、usage、error、done""" if generating.locked(): yield event("error", {"message": "上一轮还在生成，请等待完成或先停止"}) return @@ -148,14 +150,17 @@ async def send_message(body: SendIn) -> AsyncIterator[ServerSentEvent]: await save_messages(ChatMessage(session_id=current.id, role="user", content={"text": body.text})) specs = tool_specs() + effort = body.reasoning_effort or provider.reasoning_effort for round_no in range(MAX_TOOL_ROUNDS + 1): text = "" completion: llm.Completion | None = None try: - async for item in llm.stream(provider, system, messages, specs if round_no < MAX_TOOL_ROUNDS else []): + async for item in llm.stream(provider, system, messages, specs if round_no < MAX_TOOL_ROUNDS else [], effort): if isinstance(item, llm.TextDelta): text += item.text yield event("delta", {"text": item.text}) + elif isinstance(item, llm.ThinkingStarted): + yield event("thinking", {}) else: completion = item except llm.LlmError as e: @@ -170,6 +175,8 @@ async def send_message(body: SendIn) -> AsyncIterator[ServerSentEvent]: raise assistant = {"text": completion.text, "tool_calls": completion.tool_calls} + if completion.thinking_blocks: + assistant["thinking_blocks"] = completion.thinking_blocks await save_messages(ChatMessage(session_id=current.id, role="assistant", content=assistant, provider_id=provider.id, usage=completion.usage)) messages.append({"role": "assistant", **assistant}) yield event("usage", completion.usage) diff --git a/backend/src/alphapulse/api/settings.py b/backend/src/alphapulse/api/settings.py index 34b5b3b..8d81270 100644 --- a/backend/src/alphapulse/api/settings.py +++ b/backend/src/alphapulse/api/settings.py @@ -30,7 +30,7 @@ class ProviderOut(BaseModel): model: str max_tokens: int context_window: int - temperature: float | None + reasoning_effort: llm.Effort | None is_default: bool updated_at: datetime @@ -43,7 +43,7 @@ class ProviderIn(BaseModel): model: Annotated[str, Field(min_length=1, max_length=128)] max_tokens: Annotated[int, Field(ge=256, le=128000)] = 8192 context_window: Annotated[int, Field(ge=4096, le=2000000)] = 200000 - temperature: Annotated[float | None, Field(ge=0, le=2)] = None + reasoning_effort: llm.Effort | None = None is_default: bool = False @@ -56,7 +56,8 @@ class ProviderPatch(BaseModel): model: Annotated[str | None, Field(min_length=1, max_length=128)] = None max_tokens: Annotated[int | None, Field(ge=256, le=128000)] = None context_window: Annotated[int | None, Field(ge=4096, le=2000000)] = None - temperature: Annotated[float | None, Field(ge=0, le=2)] = None + # 显式传 null 表示不传思考强度 + reasoning_effort: llm.Effort | None = None is_default: bool | None = None @@ -111,9 +112,9 @@ async def create_provider(body: ProviderIn, session: Annotated[AsyncSession, Dep async def patch_provider(provider_id: int, body: ProviderPatch, session: Annotated[AsyncSession, Depends(get_session)]) -> ProviderOut: provider = await get_provider(session, provider_id) values = body.model_dump(exclude_unset=True, exclude={"api_key", "base_url", "is_default"}) - # temperature 显式传 null 表示清除，其他字段为 null 时忽略 + # reasoning_effort 显式传 null 表示清除，其他字段为 null 时忽略 for field, value in values.items(): - if value is not None or field == "temperature": + if value is not None or field == "reasoning_effort": setattr(provider, field, value) if body.base_url is not None: provider.base_url = str(body.base_url) @@ -147,7 +148,7 @@ async def test_provider(provider_id: int, session: Annotated[AsyncSession, Depen started = time.monotonic() try: completion = None - async for event in llm.stream(provider, "只回复 OK。", [{"role": "user", "text": "ping"}], []): + async for event in llm.stream(provider, "只回复 OK。", [{"role": "user", "text": "ping"}], [], provider.reasoning_effort): if isinstance(event, llm.Completion): completion = event except llm.LlmError as e: diff --git a/backend/src/alphapulse/models/ai.py b/backend/src/alphapulse/models/ai.py index d7810a8..1ee530d 100644 --- a/backend/src/alphapulse/models/ai.py +++ b/backend/src/alphapulse/models/ai.py @@ -1,6 +1,6 @@ from datetime import datetime -from sqlalchemy import BigInteger, Boolean, DateTime, Float, ForeignKey, Integer, LargeBinary, String, Text, UniqueConstraint, func +from sqlalchemy import BigInteger, Boolean, DateTime, ForeignKey, Integer, LargeBinary, String, Text, UniqueConstraint, func from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column @@ -23,7 +23,8 @@ class AiProvider(Base): max_tokens: Mapped[int] = mapped_column(Integer, default=8192) # 上下文上限（token），超过时拒绝发送而不是静默截断 context_window: Mapped[int] = mapped_column(Integer, default=200000) - temperature: Mapped[float | None] = mapped_column(Float) + # 默认思考强度；为空表示不传，由模型决定。Chat 可按轮覆盖 + reasoning_effort: Mapped[str | None] = mapped_column(String(8)) is_default: Mapped[bool] = mapped_column(Boolean, default=False) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) diff --git a/web/src/apis/settings.ts b/web/src/apis/settings.ts index b09294e..bf61e9e 100644 --- a/web/src/apis/settings.ts +++ b/web/src/apis/settings.ts @@ -2,6 +2,11 @@ import { http } from "@/http" export type Protocol = "openai" | "anthropic" +/** 思考强度，与后端 llm.Effort 一致 */ +export const EFFORTS = ["low", "medium", "high", "xhigh", "max"] as const + +export type Effort = typeof EFFORTS[number] + export interface Provider { id: number name: string @@ -11,7 +16,8 @@ export interface Provider { model: string max_tokens: number context_window: number - temperature: number | null + /** 为空表示不传，由模型决定 */ + reasoning_effort: Effort | null is_default: boolean updated_at: string } @@ -25,7 +31,8 @@ export interface ProviderForm { model: string max_tokens: number context_window: number - temperature: number | null + /** 为空表示不传，由模型决定 */ + reasoning_effort: Effort | null is_default: boolean } @@ -53,6 +60,6 @@ export function deleteProvider(id: number): Promise { } export function testProvider(id: number): Promise { - // 模型接口超时 120s 且 SDK 会重试，前端放宽等待时间 - return http.post(`/settings/providers/${id}/test`, null, { timeout: 400000 }) + // 模型接口读超时 600s 且 SDK 会重试 2 次，前端放宽等待时间 + return http.post(`/settings/providers/${id}/test`, null, { timeout: 1900000 }) } diff --git a/web/src/pages/chat/components/MessageList.vue b/web/src/pages/chat/components/MessageList.vue index c09577f..5f7c572 100644 --- a/web/src/pages/chat/components/MessageList.vue +++ b/web/src/pages/chat/components/MessageList.vue @@ -18,6 +18,8 @@ export interface DisplayItem { error?: string /** 生成中：这次模型调用已结束，后续文字另起一条 */ closed?: boolean + /** 生成中：模型正在思考，还没有输出文字 */ + thinking?: boolean } defineProps<{ items: DisplayItem[] }>() @@ -53,6 +55,9 @@ function prettyResult(result: string) {
+
+ 思考中… +
diff --git a/web/src/pages/chat/index.vue b/web/src/pages/chat/index.vue index 0872255..9102431 100644 --- a/web/src/pages/chat/index.vue +++ b/web/src/pages/chat/index.vue @@ -1,11 +1,11 @@