mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-10-23 14:49:43 +00:00
27b7df8917
Compare public commentary together with canonical function calls when recording and checking the previous response. This restores the historical full-message policy for native turns, so changed or missing commentary can distinguish otherwise identical calls. Keep canonical call history, masking, provider metadata and state advancement intact. Exclude provider IDs and separate reasoning from the comparison; identical public text and calls remain guarded. Cover commentary changes and absence, call changes and batches, provider metadata round trips, and legacy thoughts/headline behavior. Update the owning contracts. Verified with 177 passing tests, four recorded Mac warning replays and a live two-call drill using the Codex II preset.
41 lines
1.6 KiB
Python
41 lines
1.6 KiB
Python
"""Retry completed model turns that repeat the prior response."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from helpers.extension import Extension
|
|
from helpers.print_style import PrintStyle
|
|
|
|
|
|
class RepeatResponse(Extension):
|
|
def execute(self, result_data: dict[str, Any] | None = None, **kwargs: Any) -> None:
|
|
if not self.agent or not isinstance(result_data, dict):
|
|
return
|
|
if result_data.get("skip_default_processing"):
|
|
return
|
|
|
|
llm_result = result_data.get("llm_result")
|
|
response = getattr(llm_result, "response", "")
|
|
repeat_response = (
|
|
llm_result.repeat_response_text()
|
|
if getattr(llm_result, "function_calls", None) else response
|
|
)
|
|
if not isinstance(repeat_response, str) or repeat_response != self.agent.loop_data.last_response:
|
|
return
|
|
|
|
warning = self.agent.read_prompt("fw.msg_repeat.md")
|
|
log_item = self.agent.loop_data.params_temporary.get("log_item_generating")
|
|
assistant_message = self.agent.hist_add_ai_response(
|
|
response,
|
|
id=log_item.id if log_item else "",
|
|
llm_result=llm_result,
|
|
)
|
|
warning_message = self.agent.hist_add_warning(message=warning)
|
|
PrintStyle(font_color="orange", padding=True).print(warning)
|
|
self.agent.context.log.log(
|
|
type="warning",
|
|
content=f"{self.agent.agent_name}: {self.agent.read_prompt('fw.msg_repeat_response.md')}",
|
|
id=warning_message.id,
|
|
)
|
|
result_data["skip_default_processing"] = True
|