wuyang/zk-data-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: deduplicate Work execution batches

Browse Source
This commit is contained in:
wuyang
2026-07-26 14:11:57 +08:00
parent 44fe0b3dd7
commit ae5f940a5d
2 changed files with 76 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent_platform/runtime/loop.py
+29 -12
View File
@@ -489,8 +489,9 @@ class AgentLoop:
"tool_calls": tool_calls,
}
messages.append(assistant_message)
blocked_call_ids: set[str] = set()
mutation_planned = False
blocked_call_reasons: dict[str, str] = {}
batch_mutation_epoch = 0
batch_execution_epochs: dict[str, int] = {}
for call in tool_calls:
function = call.get("function") or {}
name = str(function.get("name", ""))
@@ -499,10 +500,17 @@ class AgentLoop:
except json.JSONDecodeError:
arguments = {}
if name in MUTATION_TOOLS:
mutation_planned = True
batch_mutation_epoch += 1
key = execution_call_key(name, arguments) if isinstance(arguments, dict) else None
if key and not mutation_planned and failed_execution_revisions.get(key) == mutation_count:
blocked_call_ids.add(str(call.get("id") or ""))
if not key:
continue
call_id = str(call.get("id") or "")
if failed_execution_revisions.get(key) == mutation_count and batch_mutation_epoch == 0:
blocked_call_reasons[call_id] = "failed_retry"
elif batch_execution_epochs.get(key) == batch_mutation_epoch:
blocked_call_reasons[call_id] = "batch_duplicate"
else:
batch_execution_epochs[key] = batch_mutation_epoch
results = await self._execute_calls(
calls=tool_calls,
spec=spec,
@@ -510,7 +518,7 @@ class AgentLoop:
context=tool_context,
depth=depth,
read_only=read_only,
blocked_call_ids=blocked_call_ids,
blocked_call_reasons=blocked_call_reasons,
)
execution_failures: list[str] = []
for call, result in zip(tool_calls, results, strict=True):
@@ -529,7 +537,7 @@ class AgentLoop:
if is_substantive_execution(name, arguments):
successful_executions += 1
unresolved_execution_failure = None
elif is_substantive_execution(name, arguments):
elif is_substantive_execution(name, arguments) and not result.get("policy_duplicate", False):
unresolved_execution_failure = f"{name}: {public_tool_summary(result, 1000)}"
execution_failures.append(unresolved_execution_failure)
key = execution_call_key(name, arguments) if isinstance(arguments, dict) else None
@@ -588,9 +596,9 @@ class AgentLoop:
context: ToolContext,
depth: int,
read_only: bool,
blocked_call_ids: set[str] | None = None,
blocked_call_reasons: dict[str, str] | None = None,
) -> list[dict[str, Any]]:
blocked_call_ids = blocked_call_ids or set()
blocked_call_reasons = blocked_call_reasons or {}
parsed: list[tuple[int, dict[str, Any], str, dict[str, Any], bool]] = []
for index, call in enumerate(calls):
function = call.get("function") or {}
@@ -622,7 +630,7 @@ class AgentLoop:
context=context,
depth=depth,
read_only=read_only,
blocked_repeat=str(call.get("id") or "") in blocked_call_ids,
blocked_reason=blocked_call_reasons.get(str(call.get("id") or "")),
)
index = 0
@@ -649,7 +657,7 @@ class AgentLoop:
context: ToolContext,
depth: int,
read_only: bool,
blocked_repeat: bool = False,
blocked_reason: str | None = None,
) -> dict[str, Any]:
call_id = str(call.get("id") or uuid.uuid4().hex)
public_args = {
@@ -661,7 +669,16 @@ class AgentLoop:
)
if "__parse_error__" in arguments:
result = {"ok": False, "error": arguments["__parse_error__"]}
elif blocked_repeat:
elif blocked_reason == "batch_duplicate":
result = {
"ok": False,
"policy_duplicate": True,
"error": (
"Execution policy skipped this duplicate command because the same assistant response "
"already requested it without an intervening repair."
),
}
elif blocked_reason == "failed_retry":
result = {
"ok": False,
"error": (