fix(assistant): 保留跨轮推理上下文并兼容旧会话

保留最终回答和工具前正文段的推理字段,通过已有会话存储保存并在追问时回传;按供应商、协议、接口与模型隔离上下文。兼容 DeepSeek 旧会话缺失字段,推理内容计入每轮输入预算。

验证:隔离全量后端测试 3898 项通过、13 项跳过、1 项预期失败;前端会话相关 7 项测试与生产构建通过。DeepSeek 使用协议模拟验证,未进行真实接口联调。

Signed-off-by: shy3130 <shy3130@qq.com>
This commit is contained in:
shy3130
2026-11-01 21:06:22 +08:00
parent 89e646abdd
commit f443627377
9 changed files with 324 additions and 19 deletions
+2
View File
@@ -14,6 +14,8 @@ docs/secondary-development.md 的扩展契约。
前端按钮 POST /actions/{call_id}/decision
{approve:bool}, 超时视为拒绝
{"type":"delta","content":"..."} 最终正文(M1 整段一次)
{"type":"assistant_context","reasoning_content","reasoning_context"}
隐藏的模型上下文, 随正文段保存供后续提问回传
{"type":"error","kind","message","hint?"} no_key/provider/input_too_long/rounds/model
{"type":"done"} 本轮结束(含失败)
{"type":"ping"} 心跳保活(空闲 >15s)
+23 -9
View File
@@ -28,6 +28,7 @@ from app.services.ai_provider import (
ai_configured,
current_ai_context_window,
current_ai_max_output_tokens,
current_reasoning_context_key,
is_codex_cli_provider,
)
@@ -75,13 +76,19 @@ def _classify_exception(exc: Exception) -> dict[str, str]:
return _error_event("model", text or exc.__class__.__name__)
def _truncate_history(history: list[dict[str, str]]) -> tuple[list[dict[str, str]], bool]:
def _truncate_history(
history: list[dict[str, str]], *, reasoning_context: str | None = None,
) -> tuple[list[dict[str, str]], bool]:
"""保留最近 N 条消息; 返回 (截断后历史, 是否发生了截断)。"""
clean = [
{"role": item["role"], "content": str(item.get("content") or "")}
for item in history
if item.get("role") in {"user", "assistant"} and str(item.get("content") or "").strip()
]
clean = []
for item in history:
if item.get("role") not in {"user", "assistant"} or not str(item.get("content") or "").strip():
continue
message = {"role": item["role"], "content": str(item["content"])}
if (item["role"] == "assistant" and isinstance(item.get("reasoning_content"), str)
and item.get("reasoning_context") in (None, reasoning_context)):
message["reasoning_content"] = item["reasoning_content"]
clean.append(message)
if len(clean) <= _MAX_HISTORY_MESSAGES:
return clean, False
start = len(clean) - _MAX_HISTORY_MESSAGES
@@ -97,7 +104,7 @@ def _estimate_input_tokens(messages: list[dict[str, Any]]) -> int:
"""与核心 ai_provider 相同口径的粗估: 中文 1 字 1 token, 其余 4 字符 1 token。"""
total = 0
for message in messages:
text = str(message.get("content") or "")
text = str(message.get("content") or "") + str(message.get("reasoning_content") or "")
cjk = sum(1 for ch in text if "一" <= ch <= "鿿")
total += cjk + (len(text) - cjk) // 4 + 1
return max(1, total)
@@ -166,7 +173,8 @@ async def chat_stream(
return
stream_round = stream_codex_round if is_codex_cli_provider() else stream_openai_round
truncated, was_truncated = _truncate_history(history)
reasoning_context = current_reasoning_context_key()
truncated, was_truncated = _truncate_history(history, reasoning_context=reasoning_context)
if not truncated:
yield _line(_error_event("model", "对话历史为空, 请先输入问题。"))
yield _line({"type": "done"})
@@ -261,11 +269,11 @@ async def chat_stream(
async def run() -> None:
try:
_guard_input_budget(req_messages)
text_seen = False
tool_rounds = 0
round_checkpoint = _round_checkpoint()
while True:
_guard_input_budget(req_messages)
round_text: list[str] = []
tool_calls: list[dict[str, Any]] = []
reasoning_content: str | None = None
@@ -277,6 +285,12 @@ async def chat_stream(
tool_calls = event.get("tool_calls") or []
reasoning_content = event.get("reasoning_content")
if reasoning_content is not None:
await queue.put({
"type": "assistant_context", "reasoning_content": reasoning_content,
"reasoning_context": reasoning_context,
})
if not tool_calls:
if not round_text and not text_seen:
await queue.put(_error_event("model", "本轮未产出回答, 请重试。"))
+3 -1
View File
@@ -32,6 +32,8 @@ from app.services.ndjson_heartbeat import with_heartbeat
class HistoryMessage(BaseModel):
role: Literal["user", "assistant"]
content: str = Field(min_length=1, max_length=20000)
reasoning_content: str | None = None
reasoning_context: str | None = Field(default=None, max_length=64)
class AssistantContext(BaseModel):
@@ -73,7 +75,7 @@ def build_router() -> APIRouter:
async def gen():
async for line in with_heartbeat(chat_stream(
history=[m.model_dump() for m in req.messages],
history=[m.model_dump(exclude_none=True) for m in req.messages],
context=context,
engine=engine,
data_dir=data_dir,
+12 -4
View File
@@ -13,8 +13,8 @@ tool_calls 分片, 流结束时统一返回本轮结果, 由 chat_service 决定
本轮结束; tool_calls 非空表示模型请求工具, 元素形如
{"id", "name", "arguments"(JSON 字符串)}, 上游附带 extra_content 时
(如 Gemini 的 thought_signature) 另有 "extra_content" 键, 需原样回传。
有工具调用且上游流出了思考内容 (DeepSeek 思考模式的 reasoning_content) 时
另带 "reasoning_content" 键, 下一轮必须随 assistant 消息原样回传。
上游提供了思考内容 (DeepSeek 思考模式的 reasoning_content) 时另带该键,
包括最终回答轮; 下一轮必须随 assistant 消息原样回传。
"""
from __future__ import annotations
@@ -35,6 +35,7 @@ from app.services.ai_provider import (
current_openai_reasoning_effort,
is_responses_protocol,
normalize_openai_base_url,
prepare_tool_messages,
)
# 部分推理型模型把整段预算花在 reasoning_content 上, 以 200 + length 结束。
@@ -105,12 +106,16 @@ async def stream_openai_round(
model = current_ai_model()
kwargs = _build_kwargs(temperature)
kwargs["tools"] = list(tools)
req_messages = prepare_tool_messages(
messages, model=model,
base_url=secrets_store.get_ai_config("ai_base_url", settings.ai_base_url),
)
while True:
try:
stream = await client.chat.completions.create(
model=model,
messages=[dict(m) for m in messages],
messages=req_messages,
stream=True,
**kwargs,
)
@@ -206,4 +211,7 @@ async def stream_openai_round(
if reasoning_seen:
raise RuntimeError("AI 仅返回推理内容, 未生成正文; 请检查模型配置或改用非推理模型")
raise RuntimeError("AI 服务未返回正文内容; 请检查模型配置或稍后重试")
yield {"type": "round_end", "tool_calls": [], "finish_reason": finish_reason}
end = {"type": "round_end", "tool_calls": [], "finish_reason": finish_reason}
if reasoning_field_seen:
end["reasoning_content"] = "".join(reasoning_parts)
yield end
+40 -3
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
import hashlib
import json
import logging
import os
@@ -170,6 +171,33 @@ def current_ai_model() -> str:
return current_openai_model()
def current_reasoning_context_key() -> str:
"""Bind saved reasoning to its provider, protocol, endpoint and model."""
identity = [current_ai_provider(), current_ai_protocol(), current_ai_model(),
normalize_openai_base_url(secrets_store.get_ai_config("ai_base_url", settings.ai_base_url))]
return hashlib.sha256(json.dumps(identity).encode()).hexdigest()
def prepare_tool_messages(messages: Sequence[dict], *, model: str, base_url: str) -> list[dict]:
"""Keep vendor metadata private and tolerate old DeepSeek text-only history."""
prepared = []
context_key = None
deepseek = urlsplit(base_url).hostname == "api.deepseek.com" or model.lower().rsplit("/", 1)[-1].startswith("deepseek")
for source in messages:
message = dict(source)
origin = message.pop("reasoning_context", None)
if origin is not None:
context_key = context_key or current_reasoning_context_key()
if origin != context_key:
message.pop("reasoning_content", None)
if deepseek and message.get("role") == "assistant" and message.get("reasoning_content") is None:
# Older sessions never saved reasoning; the required field can only
# be empty for those messages. Never replace a recorded empty string.
message["reasoning_content"] = ""
prepared.append(message)
return prepared
def current_openai_reasoning_effort() -> str:
stored = secrets_store.load()
if "ai_reasoning_effort" not in stored:
@@ -208,7 +236,7 @@ def _estimate_input_tokens(messages: Sequence[Message]) -> int:
"""粗略估算输入 token 数: 中文按 1 字 1 token, 其余按 4 字符 1 token。"""
total = 0
for m in messages:
text = str(m.get("content") or "")
text = str(m.get("content") or "") + str(m.get("reasoning_content") or "")
cjk = sum(1 for ch in text if "一" <= ch <= "鿿")
total += cjk + (len(text) - cjk) // 4 + 1
return max(1, total)
@@ -415,7 +443,11 @@ async def generate_ai_text_with_tools(
tool_calls = getattr(message, "tool_calls", None) or []
if not tool_calls:
# 模型不再请求工具, 产出最终文本, 结束循环。
req_messages.append({"role": "assistant", "content": message.content or ""})
assistant = {"role": "assistant", "content": message.content or ""}
reasoning = getattr(message, "reasoning_content", None)
if reasoning is not None:
assistant["reasoning_content"] = reasoning
req_messages.append(assistant)
return req_messages
assistant = {
@@ -555,7 +587,12 @@ async def _run_openai_message_once(
client = _openai_client(ai_key, timeout)
model = current_ai_model()
req_messages = list(messages)
req_messages = (
prepare_tool_messages(
messages, model=model,
base_url=secrets_store.get_ai_config("ai_base_url", settings.ai_base_url),
) if tools else list(messages)
)
kwargs = _openai_kwargs(temperature=temperature, max_tokens=max_tokens, tools=tools)
while True:
try:
@@ -7,13 +7,17 @@ thinking mode must be passed back to the API." (见 #524)。
from __future__ import annotations
import json
from types import SimpleNamespace
from typing import Any
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from openai.types.chat import ChatCompletionChunk, ChatCompletionMessage
from app.custom.assistant import chat_service, streaming
from app.custom.assistant import tools as assistant_tools
from app.custom.assistant.routes import build_router
from app.services import ai_provider
@@ -163,3 +167,149 @@ async def test_tool_loop_sends_reasoning_content_back(monkeypatch: pytest.Monkey
assert messages[1]["reasoning_content"] == reasoning
assert "reasoning_content" not in messages[3]
@pytest.mark.parametrize("reasoning", ["最终回答的推理", ""])
async def test_final_stream_round_keeps_reasoning(monkeypatch, reasoning):
events = await _events(monkeypatch, [
_chunk({"reasoning_content": reasoning}),
_chunk({"content": "完成"}, finish_reason="stop"),
])
assert events[-1]["reasoning_content"] == reasoning
@pytest.mark.parametrize("reasoning", ["最终回答的推理", ""])
async def test_nonstream_final_answer_keeps_reasoning(monkeypatch, reasoning):
async def once(*args, **kwargs):
return ChatCompletionMessage.model_validate({
"role": "assistant", "content": "完成", "reasoning_content": reasoning,
})
monkeypatch.setattr(ai_provider, "is_codex_cli_provider", lambda: False)
monkeypatch.setattr(ai_provider, "is_responses_protocol", lambda: False)
monkeypatch.setattr(ai_provider, "_check_input_budget", lambda *args, **kwargs: None)
monkeypatch.setattr(ai_provider, "_run_openai_message_once", once)
result = await ai_provider.generate_ai_text_with_tools(
[{"role": "user", "content": "hi"}], [{"type": "function", "function": {"name": "test"}}],
execute_tool=None,
)
assert result[-1]["reasoning_content"] == reasoning
@pytest.mark.parametrize("reasoning", ["最终回答的推理", ""])
def test_http_conversation_restores_reasoning_after_session_reload(monkeypatch, tmp_path, reasoning):
captured = []
class Client:
def __init__(self):
self.chat = self.completions = self
async def create(self, **kwargs):
captured.append(kwargs["messages"])
for message in kwargs["messages"]:
if message["role"] == "assistant":
assert isinstance(message.get("reasoning_content"), str)
assert "reasoning_context" not in message
if len(captured) == 1:
return _FakeStream([
_chunk({"reasoning_content": "工具推理"}),
_chunk({"tool_calls": [{"index": 0, "id": "call1", "type": "function",
"function": {"name": "list_strategies", "arguments": "{}"}}]}, "tool_calls"),
])
return _FakeStream([
_chunk({"reasoning_content": reasoning}), _chunk({"content": "完成"}, "stop"),
])
async def execute(*args, **kwargs):
return {"ok": True, "result": {}}
monkeypatch.setattr(chat_service, "ai_configured", lambda: True)
monkeypatch.setattr(chat_service, "is_codex_cli_provider", lambda: False)
monkeypatch.setattr(streaming, "is_responses_protocol", lambda: False)
monkeypatch.setattr(streaming, "_client", lambda timeout: Client())
monkeypatch.setattr(streaming, "_build_kwargs", lambda temperature: {})
monkeypatch.setattr(assistant_tools, "execute_assistant_tool", execute)
app = FastAPI()
app.include_router(build_router())
app.state.repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path))
client = TestClient(app)
user = {"role": "user", "content": "我有哪些策略"}
first = client.post("/api/custom/assistant/chat", json={"messages": [user]})
events = [json.loads(line) for line in first.text.splitlines() if line]
assert not any(e["type"] == "error" for e in events), events
metadata = [e for e in events if e["type"] == "assistant_context"][-1]
assert metadata["reasoning_content"] == reasoning
saved_message = {"id": "a", "role": "assistant", "content": "完成", "ts": 1,
"streaming": False, "reasoningContent": reasoning,
"reasoningContext": metadata["reasoning_context"]}
assert client.put("/api/custom/assistant/sessions", json={
"sessions": [{"id": "s", "messages": [saved_message]}], "activeId": "s",
}).status_code == 200
restored = client.get("/api/custom/assistant/sessions").json()["sessions"][0]["messages"][0]
second = client.post("/api/custom/assistant/chat", json={"messages": [user, {
"role": "assistant", "content": restored["content"],
"reasoning_content": restored["reasoningContent"],
"reasoning_context": restored["reasoningContext"],
}, {"role": "user", "content": "继续解释"}]})
assert not any(json.loads(line)["type"] == "error" for line in second.text.splitlines() if line)
assert captured[-1][-2]["reasoning_content"] == reasoning
def test_old_deepseek_history_gets_empty_reasoning_without_mutating_input():
messages = [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "旧回答"}]
prepared = ai_provider.prepare_tool_messages(messages, model="deepseek-flash", base_url="https://api.deepseek.com")
assert prepared[1]["reasoning_content"] == ""
assert "reasoning_content" not in messages[1]
assert ai_provider.prepare_tool_messages(messages, model="gpt-test", base_url="https://api.openai.com") == messages
def test_reasoning_from_another_provider_context_is_not_forwarded(monkeypatch):
monkeypatch.setattr(ai_provider, "current_reasoning_context_key", lambda: "current")
messages = [{"role": "assistant", "content": "回答", "reasoning_content": "old trace",
"reasoning_context": "previous"}]
prepared = ai_provider.prepare_tool_messages(messages, model="gpt-test", base_url="https://api.openai.com")
assert prepared == [{"role": "assistant", "content": "回答"}]
assert messages[0]["reasoning_content"] == "old trace"
def test_reasoning_is_included_in_input_budget(monkeypatch):
monkeypatch.setattr(ai_provider, "current_ai_context_window", lambda: 100)
monkeypatch.setattr(chat_service, "current_ai_context_window", lambda: 100)
monkeypatch.setattr(chat_service, "current_ai_max_output_tokens", lambda: 10)
messages = [{"role": "assistant", "content": "回答", "reasoning_content": "推理" * 100}]
with pytest.raises(ValueError, match="输入过长"):
ai_provider._check_input_budget(messages, max_tokens=10)
with pytest.raises(ValueError, match="输入过长"):
chat_service._guard_input_budget(messages)
def test_history_only_keeps_reasoning_for_the_matching_model_context():
history = [{"role": "assistant", "content": "回答", "reasoning_content": "trace",
"reasoning_context": "previous"}]
assert chat_service._truncate_history(history, reasoning_context="current")[0] == [
{"role": "assistant", "content": "回答"},
]
assert chat_service._truncate_history(history, reasoning_context="previous")[0][0]["reasoning_content"] == "trace"
async def test_input_budget_is_checked_after_tool_reasoning_grows(monkeypatch):
calls = []
async def round_(*args, **kwargs):
calls.append(True)
yield {"type": "round_end", "reasoning_content": "推理" * 100,
"tool_calls": [{"id": "one", "name": "list_strategies", "arguments": "{}"}]}
async def execute(*args, **kwargs):
return {"ok": True, "result": {}}
monkeypatch.setattr(chat_service, "ai_configured", lambda: True)
monkeypatch.setattr(chat_service, "is_codex_cli_provider", lambda: False)
monkeypatch.setattr(chat_service, "build_system_prompt", lambda *args, **kwargs: "s")
monkeypatch.setattr(chat_service, "current_ai_context_window", lambda: 100)
monkeypatch.setattr(chat_service, "current_ai_max_output_tokens", lambda: 10)
monkeypatch.setattr(chat_service, "stream_openai_round", round_)
monkeypatch.setattr(assistant_tools, "execute_assistant_tool", execute)
events = [json.loads(line) async for line in chat_service.chat_stream(history=[{"role": "user", "content": "hi"}])]
assert len(calls) == 1
assert next(e for e in events if e["type"] == "error")["kind"] == "input_too_long"