fix(agent): normalize fenced and flat-args LLM outputs in parser

核心算法 #10(Agent Loop):仅加固 _parse_response 解析路径——
剥除 ```json 围栏 + 收拢 action 平铺参数(deepseek 稳定输出变体,
案例 637-3/615-3 三连拒 0 步阵亡)。Thinking+JSON 协议、json_repair
兜底链、hook 时序与步数语义均未改动。
This commit is contained in:
2026-07-11 08:16:28 -04:00
parent 39c6352781
commit 8d84d5e236
2 changed files with 84 additions and 3 deletions
+27 -3
View File
@@ -18,6 +18,7 @@ TRM4 → TRM5 有意变更(非简化):
from __future__ import annotations from __future__ import annotations
import json import json
import re
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
import pluggy import pluggy
@@ -31,6 +32,9 @@ if TYPE_CHECKING:
from core.protocols import LLMProvider from core.protocols import LLMProvider
from core.types import LLMResponse from core.types import LLMResponse
# deepseek 等模型稳定输出变体:```json 围栏包裹 JSON 体
_CODE_FENCE_RE = re.compile(r"^\s*```(?:json)?\s*\n?|\n?\s*```\s*$")
async def _call_hook(hook: Any, **kwargs: Any) -> list[Any]: async def _call_hook(hook: Any, **kwargs: Any) -> list[Any]:
"""调用 pluggy hook 并 await 异步返回值。 """调用 pluggy hook 并 await 异步返回值。
@@ -267,7 +271,8 @@ class AgentLoop:
) -> tuple[str, dict, dict, str, dict, str] | None: ) -> tuple[str, dict, dict, str, dict, str] | None:
"""从 LLMResponse 中提取结构化决策数据。 """从 LLMResponse 中提取结构化决策数据。
解析流程: content → repair_json → json.loads → 校验 action/tool/args。 解析流程: content → 剥除 ```json 围栏 → repair_json → json.loads
→ 收拢 action 平铺参数 → 校验 action/tool/args。
参数: 参数:
response: LLMResponse 实例。 response: LLMResponse 实例。
@@ -282,7 +287,7 @@ class AgentLoop:
if not content.strip(): if not content.strip():
return None return None
repaired = repair_json(content) repaired = repair_json(_CODE_FENCE_RE.sub("", content).strip())
try: try:
data = json.loads(repaired) data = json.loads(repaired)
except (json.JSONDecodeError, ValueError): except (json.JSONDecodeError, ValueError):
@@ -291,7 +296,7 @@ class AgentLoop:
if not isinstance(data, dict) or "action" not in data: if not isinstance(data, dict) or "action" not in data:
return None return None
action = data["action"] action = self._normalize_action(data["action"])
if not isinstance(action, dict) or "tool" not in action or "args" not in action: if not isinstance(action, dict) or "tool" not in action or "args" not in action:
return None return None
@@ -299,6 +304,25 @@ class AgentLoop:
plan = data.get("plan", {}) plan = data.get("plan", {})
return thought, reflect, plan, content, action, response.call_id return thought, reflect, plan, content, action, response.call_id
@staticmethod
def _normalize_action(action: Any) -> Any:
"""收拢 deepseek 变体的 action 平铺参数。
deepseek 等模型稳定输出变体: 工具参数平铺在 action 下(缺 args
嵌套),确定性收拢为标准 {"tool": ..., "args": {...}} 结构。
标准嵌套结构与非法结构均原样返回,由调用方校验。
参数:
action: 从 LLM 输出解析出的 action 字段(任意类型)。
返回:
归一化后的 action(仅平铺变体被改写,其余原样返回)。
"""
if isinstance(action, dict) and "tool" in action and "args" not in action:
flat_args = {k: v for k, v in action.items() if k != "tool"}
return {"tool": action["tool"], "args": flat_args}
return action
async def _execute_tool( async def _execute_tool(
self, self,
dispatcher: ToolDispatcher, dispatcher: ToolDispatcher,
+57
View File
@@ -239,3 +239,60 @@ class TestAgentLoop:
assert "after_tool:0" in tracker.events assert "after_tool:0" in tracker.events
assert "after_step:0" in tracker.events assert "after_step:0" in tracker.events
assert "on_finish:finished" in tracker.events assert "on_finish:finished" in tracker.events
# ── A1 解析容错测试(Spec-1)──────────────────────────────────
# 生产真实样本结构:尾部围栏残留 + action.args 平铺(开头围栏场景由
# test_leading_json_fence 单独覆盖)
_REAL_FLAT_FENCED = """{
"plan": {
"goal": "从三个L1根节点开始建立全局认知",
"tool": "view_node",
"reason": "三个L1节点覆盖整个视频"
},
"action": {
"tool": "view_node",
"node_id": "J5Npf2xJpag_L1_000",
"question": "What is the overall topic of this video?"
}
}
```"""
class TestParseNormalization:
"""deepseek 输出变体(args 平铺 + ```json 围栏)归一化。"""
def _parse(self, content: str):
loop = AgentLoop(llm=AsyncMock(), max_steps=10)
return loop._parse_response(_make_response(content))
def test_flat_args_with_trailing_fence(self) -> None:
"""生产样本:action 平铺 node_id/question + 尾部围栏。"""
parsed = self._parse(_REAL_FLAT_FENCED)
assert parsed is not None
action = parsed[4]
assert action["tool"] == "view_node"
assert action["args"] == {
"node_id": "J5Npf2xJpag_L1_000",
"question": "What is the overall topic of this video?",
}
def test_leading_json_fence(self) -> None:
content = '```json\n{"reflect": {}, "plan": {}, "action": {"tool": "submit_answer", "args": {"answer": "A"}}}\n```'
parsed = self._parse(content)
assert parsed is not None
assert parsed[4]["args"] == {"answer": "A"}
def test_nested_args_unchanged(self) -> None:
"""标准嵌套结构不受归一化影响。"""
parsed = self._parse(_submit_json("B"))
assert parsed is not None
assert parsed[4] == {"tool": "submit_answer", "args": {"answer": "B"}}
def test_action_missing_tool_still_rejected(self) -> None:
content = json.dumps({"reflect": {}, "plan": {}, "action": {"node_id": "x"}})
assert self._parse(content) is None
def test_empty_content_still_rejected(self) -> None:
assert self._parse("") is None