Detect truncated model output and skip truncated tool calls
This commit is contained in:
1 parent
5b7490ac48
commit
438d6a00d9
2 files changed
+20
-5
No files matched your search
@@ -54,6 +54,8 @@ class Completion:
|
|||||||
usage: dict = field(default_factory=dict)
|
usage: dict = field(default_factory=dict)
|
||||||
# Anthropic 的 thinking / redacted_thinking 块:工具调用循环中必须原样回传,否则接口报错
|
# Anthropic 的 thinking / redacted_thinking 块:工具调用循环中必须原样回传,否则接口报错
|
||||||
thinking_blocks: list[dict] = field(default_factory=list)
|
thinking_blocks: list[dict] = field(default_factory=list)
|
||||||
|
# 达到 max_tokens 被截断:工具调用的参数可能不完整,不能执行
|
||||||
|
truncated: bool = False
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@@ -175,6 +177,8 @@ async def _stream_openai(provider: AiProvider, system: str, messages: list[dict]
|
|||||||
result.usage = {"input_tokens": chunk.usage.prompt_tokens, "output_tokens": chunk.usage.completion_tokens}
|
result.usage = {"input_tokens": chunk.usage.prompt_tokens, "output_tokens": chunk.usage.completion_tokens}
|
||||||
if not chunk.choices:
|
if not chunk.choices:
|
||||||
continue
|
continue
|
||||||
|
if chunk.choices[0].finish_reason == "length":
|
||||||
|
result.truncated = True
|
||||||
delta = chunk.choices[0].delta
|
delta = chunk.choices[0].delta
|
||||||
if delta.content:
|
if delta.content:
|
||||||
result.text += delta.content
|
result.text += delta.content
|
||||||
@@ -187,7 +191,7 @@ async def _stream_openai(provider: AiProvider, system: str, messages: list[dict]
|
|||||||
call["name"] += tc.function.name
|
call["name"] += tc.function.name
|
||||||
if tc.function and tc.function.arguments:
|
if tc.function and tc.function.arguments:
|
||||||
call["arguments"] += tc.function.arguments
|
call["arguments"] += tc.function.arguments
|
||||||
result.tool_calls = [_parse_call(c["id"], c["name"], c["arguments"]) for _, c in sorted(calls.items())]
|
result.tool_calls = [_parse_call(c["id"], c["name"], c["arguments"], result.truncated) for _, c in sorted(calls.items())]
|
||||||
yield result
|
yield result
|
||||||
|
|
||||||
|
|
||||||
@@ -260,12 +264,16 @@ async def _stream_anthropic(provider: AiProvider, system: str, messages: list[di
|
|||||||
thinking[event.index]["signature"] += event.delta.signature
|
thinking[event.index]["signature"] += event.delta.signature
|
||||||
elif event.type == "message_delta":
|
elif event.type == "message_delta":
|
||||||
result.usage["output_tokens"] = event.usage.output_tokens
|
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.truncated = event.delta.stop_reason == "max_tokens"
|
||||||
|
result.tool_calls = [_parse_call(b["id"], b["name"], b["arguments"], result.truncated) for _, b in sorted(blocks.items())]
|
||||||
result.thinking_blocks = [b for _, b in sorted(thinking.items())]
|
result.thinking_blocks = [b for _, b in sorted(thinking.items())]
|
||||||
yield result
|
yield result
|
||||||
|
|
||||||
|
|
||||||
def _parse_call(call_id: str, name: str, arguments: str) -> dict:
|
def _parse_call(call_id: str, name: str, arguments: str, truncated: bool) -> dict:
|
||||||
|
if truncated:
|
||||||
|
# 截断的参数是半截 JSON,不解析;调用方不会执行它
|
||||||
|
return {"id": call_id, "name": name, "arguments": {}}
|
||||||
try:
|
try:
|
||||||
parsed = json.loads(arguments) if arguments else {}
|
parsed = json.loads(arguments) if arguments else {}
|
||||||
except json.JSONDecodeError as e:
|
except json.JSONDecodeError as e:
|
||||||
|
|||||||
@@ -15,7 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
|||||||
|
|
||||||
from alphapulse.ai import llm
|
from alphapulse.ai import llm
|
||||||
from alphapulse.ai.prompt import build_system_prompt, current_context, estimate_tokens
|
from alphapulse.ai.prompt import build_system_prompt, current_context, estimate_tokens
|
||||||
from alphapulse.ai.tools import run_tool, tool_specs
|
from alphapulse.ai.tools import ToolOutcome, run_tool, tool_specs
|
||||||
from alphapulse.core.auth import require_admin
|
from alphapulse.core.auth import require_admin
|
||||||
from alphapulse.core.db import SessionLocal, get_session
|
from alphapulse.core.db import SessionLocal, get_session
|
||||||
from alphapulse.models import AiProvider, ChatMessage, ChatSession, ContextVersion
|
from alphapulse.models import AiProvider, ChatMessage, ChatSession, ContextVersion
|
||||||
@@ -27,6 +27,7 @@ MAX_TOOL_ROUNDS = 8
|
|||||||
# 发给前端展示的工具结果长度
|
# 发给前端展示的工具结果长度
|
||||||
TOOL_PREVIEW_CHARS = 2000
|
TOOL_PREVIEW_CHARS = 2000
|
||||||
STOPPED_RESULT = json.dumps({"error": "用户已停止生成,工具未执行"}, ensure_ascii=False)
|
STOPPED_RESULT = json.dumps({"error": "用户已停止生成,工具未执行"}, ensure_ascii=False)
|
||||||
|
TRUNCATED_RESULT = json.dumps({"error": "输出达到最大输出上限被截断,工具参数不完整,未执行。请精简内容后重新调用"}, ensure_ascii=False)
|
||||||
|
|
||||||
# api 只有一个进程:同一时间只允许一轮生成,Reset 也要等生成结束
|
# api 只有一个进程:同一时间只允许一轮生成,Reset 也要等生成结束
|
||||||
generating = asyncio.Lock()
|
generating = asyncio.Lock()
|
||||||
@@ -177,10 +178,15 @@ async def send_message(body: SendIn) -> AsyncIterator[ServerSentEvent]:
|
|||||||
assistant = {"text": completion.text, "tool_calls": completion.tool_calls}
|
assistant = {"text": completion.text, "tool_calls": completion.tool_calls}
|
||||||
if completion.thinking_blocks:
|
if completion.thinking_blocks:
|
||||||
assistant["thinking_blocks"] = completion.thinking_blocks
|
assistant["thinking_blocks"] = completion.thinking_blocks
|
||||||
|
truncated_message = f"模型输出达到最大输出上限({provider.max_tokens} token)被截断,可在「环境变量」页调大「最大输出」"
|
||||||
|
if completion.truncated and not completion.tool_calls:
|
||||||
|
assistant["error"] = truncated_message
|
||||||
await save_messages(ChatMessage(session_id=current.id, role="assistant", content=assistant, provider_id=provider.id, usage=completion.usage))
|
await save_messages(ChatMessage(session_id=current.id, role="assistant", content=assistant, provider_id=provider.id, usage=completion.usage))
|
||||||
messages.append({"role": "assistant", **assistant})
|
messages.append({"role": "assistant", **assistant})
|
||||||
yield event("usage", completion.usage)
|
yield event("usage", completion.usage)
|
||||||
if not completion.tool_calls:
|
if not completion.tool_calls:
|
||||||
|
if completion.truncated:
|
||||||
|
yield event("error", {"message": truncated_message})
|
||||||
break
|
break
|
||||||
|
|
||||||
pending = list(completion.tool_calls)
|
pending = list(completion.tool_calls)
|
||||||
@@ -188,7 +194,8 @@ async def send_message(body: SendIn) -> AsyncIterator[ServerSentEvent]:
|
|||||||
while pending:
|
while pending:
|
||||||
call = pending[0]
|
call = pending[0]
|
||||||
yield event("tool_call", {"id": call["id"], "name": call["name"], "arguments": call["arguments"]})
|
yield event("tool_call", {"id": call["id"], "name": call["name"], "arguments": call["arguments"]})
|
||||||
outcome = await run_tool(call["name"], call["arguments"])
|
# 截断时不执行,把原因告诉模型,让它精简后重试
|
||||||
|
outcome = ToolOutcome(TRUNCATED_RESULT) if completion.truncated else await run_tool(call["name"], call["arguments"])
|
||||||
result = {"tool_call_id": call["id"], "name": call["name"], "result": outcome.result}
|
result = {"tool_call_id": call["id"], "name": call["name"], "result": outcome.result}
|
||||||
await save_messages(ChatMessage(session_id=current.id, role="tool", content=result))
|
await save_messages(ChatMessage(session_id=current.id, role="tool", content=result))
|
||||||
pending.pop(0)
|
pending.pop(0)
|
||||||
|
|||||||
Reference in new issue
Block a user