diff --git a/backend/app/services/ai_provider.py b/backend/app/services/ai_provider.py index 1fa1e78..e7aae8c 100644 --- a/backend/app/services/ai_provider.py +++ b/backend/app/services/ai_provider.py @@ -72,8 +72,8 @@ _ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]") # ---------------------------------------------------------------- -# 用户 focus 输入净化 — 防止通过"特别关注"绕过红线诱导 AI 给出买卖建议 -# 命中任一敏感词时,整个 focus 被丢弃(返回空串),由各 analyzer 据此跳过注入。 +# 用户 focus 输入规范化。交易建议类表达不会被静默丢弃,而是由统一提示词 +# 转换成客观价位、风险和情景分析,避免历史报告显示了 focus、模型却没有收到。 # ---------------------------------------------------------------- _FOCUS_BLOCKLIST = re.compile( r"买入|卖出|加仓|减仓|轻仓|重仓|半仓|全仓|仓位|止损|止盈|" @@ -87,19 +87,36 @@ _FOCUS_BLOCKLIST = re.compile( def sanitize_focus(focus: str) -> str: - """净化用户输入的 focus 文本。 - - 命中交易指令/投资建议类敏感词时返回空串,阻止其注入 AI 提示词。 - 这是对系统提示词红线的兜底:即便用户试图通过 focus 绕过,也不会生效。 - """ + """规范化 focus 中的首尾空白与连续换行。""" if not focus: return "" - text = focus.strip() + text = re.sub(r"\s+", " ", focus).strip() + return text + + +def build_focus_instruction(focus: str, *, report_name: str = "分析报告") -> str: + """构建所有报告共用的关注重点指令。 + + 有关注点时要求模型在固定报告结构之前先直接回应。若原问题涉及交易 + 建议,保留问题语义但要求转换成中立的数据分析,不再无提示地整段丢弃。 + """ + text = sanitize_focus(focus) if not text: return "" + + lines = [ + "## 用户关注重点(必须优先回应)", + f"用户关注: {text}", + f"请在完整{report_name}最前面先输出 `### 0. 🔎 关注重点回应`," + "用 2-4 条带具体数据的结论直接回应;随后继续完成既定报告结构," + "并在相关章节加深分析。不要只复述问题。", + ] if _FOCUS_BLOCKLIST.search(text): - return "" - return text + lines.append( + "该关注点含有买卖、仓位、目标价或预测类表达。不得给出相应操作结论;" + "请将其转换为客观的技术/财务状态、关键价位、风险因素和条件情景后回应。" + ) + return "\n".join(lines) def current_ai_provider() -> str: @@ -309,11 +326,13 @@ async def stream_ai_text( temperature: float | None = 0.5, max_tokens: int | None = 4000, timeout: float = 180.0, + prefer_final_answer: bool = False, ) -> AsyncIterator[str]: """Yield text deltas from the configured provider. Codex CLI only exposes the final assistant message for this use case, so it - yields one complete chunk after the command exits. + yields one complete chunk after the command exits. ``prefer_final_answer`` + lets compatible providers prioritize visible content over hidden reasoning. max_tokens=None 表示不限制输出(同 generate_ai_text 的说明)。 """ @@ -328,6 +347,7 @@ async def stream_ai_text( temperature=temperature, max_tokens=max_tokens, timeout=timeout, + prefer_final_answer=prefer_final_answer, ): yield chunk @@ -374,6 +394,7 @@ async def _stream_openai( temperature: float | None, max_tokens: int | None, timeout: float, + prefer_final_answer: bool, ) -> AsyncIterator[str]: ai_key = secrets_store.get_ai_key() if not ai_key: @@ -381,15 +402,16 @@ async def _stream_openai( client = _openai_client(ai_key, timeout) model = current_ai_model() + base_url = secrets_store.get_ai_config("ai_base_url", settings.ai_base_url) req_messages = list(messages) - async def _iter(stream): - async for chunk in stream: - delta = chunk.choices[0].delta if chunk.choices else None - if delta and delta.content: - yield delta.content - - kwargs = _openai_kwargs(temperature=temperature, max_tokens=max_tokens) + kwargs = _openai_kwargs( + temperature=temperature, + max_tokens=max_tokens, + model=model, + base_url=base_url, + prefer_final_answer=prefer_final_answer, + ) while True: try: stream = await client.chat.completions.create( @@ -410,7 +432,7 @@ async def _stream_openai( raise try: - async for piece in _iter(stream): + async for piece in _iter_openai_text(stream): yield piece except Exception as exc: if _is_openai_transport_error(exc): @@ -418,6 +440,53 @@ async def _stream_openai( raise +_LENGTH_FINISH_REASONS = {"length", "max_tokens", "max_output_tokens"} + + +async def _iter_openai_text(stream) -> AsyncIterator[str]: + """Normalize an OpenAI-compatible stream into complete text deltas. + + Reasoning models may spend the entire completion budget on + ``reasoning_content`` and finish with HTTP 200 but no user-visible text. + Treat that response, and any length-truncated partial response, as a + terminal generation error instead of silently reporting success. + """ + content_seen = False + reasoning_seen = False + finish_reason = "" + + async for chunk in stream: + choices = getattr(chunk, "choices", None) or [] + if not choices: + continue + choice = choices[0] + reason = getattr(choice, "finish_reason", None) + if reason: + finish_reason = str(reason) + + delta = getattr(choice, "delta", None) + if delta is None: + continue + if getattr(delta, "reasoning_content", None): + reasoning_seen = True + content = getattr(delta, "content", None) + if content: + content_seen = True + yield content + + if finish_reason in _LENGTH_FINISH_REASONS: + if reasoning_seen and not content_seen: + raise RuntimeError( + "AI 推理达到输出长度上限, 未生成正文; 请提高输出 Token 上限或改用非推理模型" + ) + raise RuntimeError("AI 输出达到长度上限, 内容不完整; 请提高输出 Token 上限后重试") + + if not content_seen: + if reasoning_seen: + raise RuntimeError("AI 仅返回推理内容, 未生成正文; 请检查模型配置或改用非推理模型") + raise RuntimeError("AI 服务未返回正文内容; 请检查模型配置或稍后重试") + + def _openai_client(api_key: str, timeout: float): from openai import AsyncOpenAI @@ -479,8 +548,18 @@ def _openai_retry_kwargs(exc: Exception, kwargs: dict) -> dict | None: return None -def _openai_kwargs(*, temperature: float | None, max_tokens: int | None) -> dict: - """Build OpenAI create() kwargs; optional parameters are omitted when empty. +_DEEPSEEK_V4_MODELS = {"deepseek-v4-flash", "deepseek-v4-pro"} + + +def _openai_kwargs( + *, + temperature: float | None, + max_tokens: int | None, + model: str = "", + base_url: str = "", + prefer_final_answer: bool = False, +) -> dict: + """Build OpenAI create() kwargs and map supported provider capabilities. max_tokens=None 时不传 — 由服务端默认上限管理(推理模型的思考 token 也 计入该参数预算, 限制会挤占正文, 见 stream_ai_text 文档)。 @@ -494,6 +573,15 @@ def _openai_kwargs(*, temperature: float | None, max_tokens: int | None) -> dict reasoning_effort = current_openai_reasoning_effort() if reasoning_effort: kwargs["reasoning_effort"] = reasoning_effort + if ( + prefer_final_answer + and model.strip().lower() in _DEEPSEEK_V4_MODELS + and urlsplit(base_url.strip()).hostname == "api.deepseek.com" + ): + # DeepSeek V4 defaults to thinking mode. For report-style tasks the + # hidden reasoning shares max_tokens with the final answer and can + # exhaust the budget before any visible content is emitted. + kwargs["extra_body"] = {"thinking": {"type": "disabled"}} return kwargs diff --git a/backend/app/services/concept_rotation_analyzer.py b/backend/app/services/concept_rotation_analyzer.py index 9dbbdc5..79b00b6 100644 --- a/backend/app/services/concept_rotation_analyzer.py +++ b/backend/app/services/concept_rotation_analyzer.py @@ -278,10 +278,10 @@ def _build_user_prompt(signals: dict, overview: dict, days: int, dates: list[str _build_signal_block("🎰 游资特征 (排名波动大)", signals.get("hot_money", [])), ] - from app.services.ai_provider import sanitize_focus - safe_focus = sanitize_focus(focus) - if safe_focus: - parts.extend(["", f"本次分析请特别关注: {safe_focus}"]) + from app.services.ai_provider import build_focus_instruction + focus_instruction = build_focus_instruction(focus, report_name=f"{dim}轮动分析报告") + if focus_instruction: + parts.extend(["", focus_instruction]) return "\n".join(parts) @@ -374,6 +374,7 @@ async def analyze_rotation_stream( temperature=0.5, # 不限制输出(推理模型思考 token 计入预算, 见 ai_provider.stream_ai_text) max_tokens=None, + prefer_final_answer=True, ): got_content = True yield json.dumps({"type": "delta", "content": delta}, ensure_ascii=False) diff --git a/backend/app/services/financial_analyzer.py b/backend/app/services/financial_analyzer.py index 494f53f..f3edfb4 100644 --- a/backend/app/services/financial_analyzer.py +++ b/backend/app/services/financial_analyzer.py @@ -136,13 +136,10 @@ def _build_user_prompt(fins: dict[str, list[dict]], symbol: str, focus: str) -> data_json, "```", ] - from app.services.ai_provider import sanitize_focus - safe_focus = sanitize_focus(focus) - if safe_focus: - lines.extend([ - "", - f"本次分析请特别关注: {safe_focus}", - ]) + from app.services.ai_provider import build_focus_instruction + focus_instruction = build_focus_instruction(focus, report_name="财务分析报告") + if focus_instruction: + lines.extend(["", focus_instruction]) return "\n".join(lines) @@ -187,6 +184,7 @@ async def analyze_financials_stream( temperature=0.4, # 不限制输出(推理模型思考 token 计入预算, 见 ai_provider.stream_ai_text) max_tokens=None, + prefer_final_answer=True, ): got_content = True yield json.dumps({"type": "delta", "content": delta}, ensure_ascii=False) diff --git a/backend/app/services/market_recap.py b/backend/app/services/market_recap.py index 9be3bb7..5cf5ee0 100644 --- a/backend/app/services/market_recap.py +++ b/backend/app/services/market_recap.py @@ -237,10 +237,10 @@ def _build_user_prompt(overview: dict, news: list[dict], focus: str, lhb_context "消息催化一节请直接从量价异动给出可能的催化逻辑结论,不要编造具体消息,也不要复述本说明。)", ]) - from app.services.ai_provider import sanitize_focus - safe_focus = sanitize_focus(focus) - if safe_focus: - parts.extend(["", f"本次复盘请特别关注: {safe_focus}"]) + from app.services.ai_provider import build_focus_instruction + focus_instruction = build_focus_instruction(focus, report_name="大盘复盘报告") + if focus_instruction: + parts.extend(["", focus_instruction]) return "\n".join(parts) @@ -336,6 +336,7 @@ async def recap_market_stream( temperature=0.5, # 不限制输出(推理模型思考 token 计入预算, 见 ai_provider.stream_ai_text) max_tokens=None, + prefer_final_answer=True, ): got_content = True yield json.dumps({"type": "delta", "content": delta}, ensure_ascii=False) diff --git a/backend/app/services/stock_analyzer.py b/backend/app/services/stock_analyzer.py index 949054e..3aa404b 100644 --- a/backend/app/services/stock_analyzer.py +++ b/backend/app/services/stock_analyzer.py @@ -16,8 +16,8 @@ from __future__ import annotations import json import logging +from collections.abc import AsyncIterator from pathlib import Path -from typing import AsyncIterator import polars as pl @@ -238,10 +238,10 @@ def _build_user_prompt( "请按系统提示词第 4 节的说明,在基本面/财务面维度给出\"接入中\"的友好提示,不要编造数据。)", ]) - from app.services.ai_provider import sanitize_focus - safe_focus = sanitize_focus(focus) - if safe_focus: - parts.extend(["", f"本次分析请特别关注: {safe_focus}"]) + from app.services.ai_provider import build_focus_instruction + focus_instruction = build_focus_instruction(focus, report_name="个股分析报告") + if focus_instruction: + parts.extend(["", focus_instruction]) return "\n".join(parts) @@ -325,11 +325,12 @@ async def analyze_stock_stream( # 不限制输出: 推理模型(deepseek reasoner 系)思考 token 计入 max_tokens # 预算, 固定上限会把正文挤光(实测 4500 全被推理吃掉 → 正文 0 字)。 max_tokens=None, + prefer_final_answer=True, ): got_content = True yield json.dumps({"type": "delta", "content": delta}, ensure_ascii=False) - except Exception as e: # noqa: BLE001 + except Exception as e: logger.exception("AI stock analysis failed for %s: %s", symbol, e) yield json.dumps({"type": "error", "message": f"AI 分析失败: {e}"}, ensure_ascii=False) return diff --git a/backend/tests/test_ai_analysis_focus.py b/backend/tests/test_ai_analysis_focus.py new file mode 100644 index 0000000..8a63bb1 --- /dev/null +++ b/backend/tests/test_ai_analysis_focus.py @@ -0,0 +1,62 @@ +"""AI 分析关注重点的统一 Prompt 契约测试。""" + +from app.services.ai_provider import build_focus_instruction, sanitize_focus +from app.services.concept_rotation_analyzer import ( + _build_user_prompt as build_rotation_user_prompt, +) +from app.services.financial_analyzer import ( + _build_user_prompt as build_financial_user_prompt, +) +from app.services.market_recap import ( + _build_user_prompt as build_recap_user_prompt, +) +from app.services.stock_analyzer import ( + _build_user_prompt as build_stock_user_prompt, +) + + +def test_focus_whitespace_is_normalized() -> None: + assert sanitize_focus(" 支撑位\n 多少 ") == "支撑位 多少" + + +def test_trade_wording_is_reframed_instead_of_silently_dropped() -> None: + instruction = build_focus_instruction("现在能买吗,目标价多少", report_name="个股分析报告") + + assert "用户关注: 现在能买吗,目标价多少" in instruction + assert "关注重点回应" in instruction + assert "不得给出相应操作结论" in instruction + assert "关键价位" in instruction + + +def test_safe_focus_requires_a_direct_answer_without_extra_warning() -> None: + instruction = build_focus_instruction("支撑位多少", report_name="个股分析报告") + + assert "用户关注: 支撑位多少" in instruction + assert "用 2-4 条带具体数据的结论直接回应" in instruction + assert "不得给出相应操作结论" not in instruction + + +def test_all_focus_enabled_analyzers_share_the_priority_contract() -> None: + focus = "支撑位多少" + prompts = [ + build_stock_user_prompt([], {}, {}, 10.0, "600000.SH", focus), + build_financial_user_prompt({}, "600000.SH", focus), + build_recap_user_prompt({}, [], focus), + build_rotation_user_prompt({}, {}, 12, [], focus), + ] + + for prompt in prompts: + assert "## 用户关注重点(必须优先回应)" in prompt + assert f"用户关注: {focus}" in prompt + assert "### 0. 🔎 关注重点回应" in prompt + + +def test_empty_focus_does_not_add_focus_section() -> None: + prompts = [ + build_stock_user_prompt([], {}, {}, 10.0, "600000.SH", ""), + build_financial_user_prompt({}, "600000.SH", ""), + build_recap_user_prompt({}, [], ""), + build_rotation_user_prompt({}, {}, 12, [], ""), + ] + + assert all("用户关注重点" not in prompt for prompt in prompts) diff --git a/backend/tests/test_ai_provider.py b/backend/tests/test_ai_provider.py index 65d9eeb..7995e56 100644 --- a/backend/tests/test_ai_provider.py +++ b/backend/tests/test_ai_provider.py @@ -1,6 +1,7 @@ from __future__ import annotations import tomllib +from types import SimpleNamespace import httpx import openai @@ -48,6 +49,41 @@ def test_normalize_openai_base_url_strips_trailing_slash(): assert normalize_openai_base_url("https://open.bigmodel.cn/api/paas/v4/") == "https://open.bigmodel.cn/api/paas/v4" +def test_openai_kwargs_prefers_final_answer_for_official_deepseek_v4(): + kwargs = ai_provider._openai_kwargs( + temperature=0.5, + max_tokens=8192, + model="deepseek-v4-pro", + base_url="https://api.deepseek.com/v1", + prefer_final_answer=True, + ) + + assert kwargs["extra_body"] == {"thinking": {"type": "disabled"}} + + +def test_openai_kwargs_does_not_send_deepseek_option_to_other_providers(): + kwargs = ai_provider._openai_kwargs( + temperature=0.5, + max_tokens=8192, + model="gpt-5.5", + base_url="https://api.openai.com/v1", + prefer_final_answer=True, + ) + + assert "extra_body" not in kwargs + + +def test_openai_kwargs_keeps_deepseek_default_without_final_answer_preference(): + kwargs = ai_provider._openai_kwargs( + temperature=0.5, + max_tokens=8192, + model="deepseek-v4-pro", + base_url="https://api.deepseek.com/v1", + ) + + assert "extra_body" not in kwargs + + def test_format_openai_error_hides_html_gateway_body(): response = httpx.Response( 504, @@ -400,6 +436,67 @@ def test_save_ai_settings_rejects_non_positive(monkeypatch): settings_api.save_ai_settings(req2) +async def _fake_openai_stream(*chunks): + for chunk in chunks: + yield chunk + + +def _stream_chunk(*, content=None, reasoning_content=None, finish_reason=None): + delta = SimpleNamespace(content=content, reasoning_content=reasoning_content) + choice = SimpleNamespace(delta=delta, finish_reason=finish_reason) + return SimpleNamespace(choices=[choice]) + + +@pytest.mark.asyncio +async def test_iter_openai_text_rejects_reasoning_only_length_exhaustion(): + stream = _fake_openai_stream( + _stream_chunk(reasoning_content="内部推理"), + _stream_chunk(finish_reason="length"), + ) + + with pytest.raises(RuntimeError, match="推理达到输出长度上限"): + async for _ in ai_provider._iter_openai_text(stream): + pass + + +@pytest.mark.asyncio +async def test_iter_openai_text_rejects_truncated_partial_content(): + stream = _fake_openai_stream( + _stream_chunk(content="未完成正文"), + _stream_chunk(finish_reason="length"), + ) + pieces = [] + + with pytest.raises(RuntimeError, match="输出达到长度上限"): + async for piece in ai_provider._iter_openai_text(stream): + pieces.append(piece) + + assert pieces == ["未完成正文"] + + +@pytest.mark.asyncio +async def test_iter_openai_text_yields_complete_content_and_ignores_reasoning(): + stream = _fake_openai_stream( + _stream_chunk(reasoning_content="内部推理"), + _stream_chunk(content="完整"), + _stream_chunk(content="正文"), + _stream_chunk(finish_reason="stop"), + ) + + pieces = [piece async for piece in ai_provider._iter_openai_text(stream)] + + assert pieces == ["完整", "正文"] + + +@pytest.mark.asyncio +async def test_iter_openai_text_rejects_stream_without_content(): + stream = _fake_openai_stream(_stream_chunk(finish_reason="stop")) + + with pytest.raises(RuntimeError, match="未返回正文内容"): + async for _ in ai_provider._iter_openai_text(stream): + pass + + def test_codex_process_env_excludes_application_secrets(monkeypatch, tmp_path): monkeypatch.setenv("PATH", "test-path") monkeypatch.setenv("HTTPS_PROXY", "http://proxy.example")