diff --git a/app/harness/inference.py b/app/harness/inference.py index 6092025..911dde5 100644 --- a/app/harness/inference.py +++ b/app/harness/inference.py @@ -14,6 +14,7 @@ from __future__ import annotations import asyncio import json +import sqlite3 from collections import defaultdict from dataclasses import dataclass from typing import TYPE_CHECKING, Any @@ -162,6 +163,24 @@ def _to_text_field(value: Any) -> str: return json.dumps(value, ensure_ascii=False) +def _normalize_prediction(answer: object) -> str | None: + """归一化 prediction 落库值。 + + LLM 提交的 answer 有时是 list/dict(如 {'answer': ['B']}),sqlite 无法绑定 + 非标量类型直接入库会抛 ProgrammingError 击穿整轮 gather。None 保留(INFRA 空 + 预测语义,供正确率判定天然计错);str 原样;其余 JSON 序列化为文本。 + + 参数: + answer: LoopResult.result 中的 answer 原始值(可能是 None/str/list/dict)。 + + 返回: + None(保留空预测语义)或可直接入库的字符串。 + """ + if answer is None or isinstance(answer, str): + return answer + return _to_text_field(answer) + + def _zero_result(run_id: str) -> InferenceResult: """空记录时的零值 InferenceResult。 @@ -419,7 +438,7 @@ async def _run_single_question( reasoning = _to_text_field(result_dict.get("reasoning", "")) record.update( { - "prediction": result_dict.get("answer"), + "prediction": _normalize_prediction(result_dict.get("answer")), "evidence": evidence, "reasoning": reasoning, "steps_used": loop_result.steps_used, @@ -442,8 +461,18 @@ async def _run_single_question( except Exception: logger.exception("[{}] QA {} 执行异常", qa.video_id, qa.question_id) - # prediction 必落库(try 外,无论成败) - await asyncio.to_thread(log.insert, "predictions", record) + # prediction 必落库(try 外,无论成败);绑定异常降级为最小 error 行,不击穿 gather + try: + await asyncio.to_thread(log.insert, "predictions", record) + except (sqlite3.InterfaceError, sqlite3.ProgrammingError): + logger.exception("[{}] QA {} 落库绑定异常,降级为 error 行", qa.video_id, qa.question_id) + record["prediction"] = None + record["stop_reason"] = "error" + await asyncio.to_thread( + log.insert, + "predictions", + {k: v for k, v in record.items() if isinstance(v, (str, int, float, type(None)))}, + ) return record diff --git a/tests/unit/test_harness_inference.py b/tests/unit/test_harness_inference.py index 2ec702e..fd57b04 100644 --- a/tests/unit/test_harness_inference.py +++ b/tests/unit/test_harness_inference.py @@ -559,6 +559,65 @@ class TestPredictionAlwaysWritten: assert rows[0]["prediction"] is None +def _make_nonscalar_llm_response() -> LLMResponse: + """构造 submit_answer 提交非标量 answer(list)的 LLMResponse。""" + content = json.dumps( + { + "reflect": {"observation": "找到答案"}, + "plan": {"next_step": "提交"}, + "action": { + "tool": "submit_answer", + "args": { + "answer": ["B"], + "evidence": "证据文本", + "reasoning": "推理过程", + }, + }, + } + ) + return LLMResponse( + content=content, + thinking="思考过程", + model="test-model", + provider="test", + prompt_tokens=100, + completion_tokens=50, + latency_ms=200, + ttft_ms=30.0, + max_inter_token_ms=5.0, + cache_hit=False, + call_id="test-call-nonscalar", + ) + + +class TestNonScalarPrediction: + """非标量 prediction 归一化 + 落库加固测试。""" + + @pytest.mark.asyncio + async def test_nonscalar_prediction_does_not_crash(self, harness_log: HarnessLog) -> None: + """submit_answer 返回 {'answer': ['B']} 时归一化落库,不抛 sqlite 绑定异常。""" + llm = AsyncMock() + llm.chat.return_value = _make_nonscalar_llm_response() + + result = await run_inference( + [_make_question(answer="B")], + llm=llm, + tool_dispatch_fn=_stub_tool_dispatch, + prompt_builder=_stub_prompt_builder, + log=harness_log, + run_id="run-nonscalar", + concurrency=1, + max_steps=10, + skill_mode="auto", + ) + + assert result.total == 1 + rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",)) + assert len(rows) == 1 + # prediction 被 JSON 序列化为字符串,不再是 Python list + assert rows[0]["prediction"] == '["B"]' + + class TestPluginsFactory: """plugins_factory 调用测试。"""