Fimeg/letta-server
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat: Factor our approval parsing logic (#4782)

Browse Source 
Factor our approval parsing logic
This commit is contained in:
Matthew Zhou
2025-09-18 15:05:49 -07:00
committed by Caren Thomas
parent c85bfefa52
commit 41e1f01659
2 changed files with 10 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
letta/agents/helpers.py
View File

@@ -282,3 +282,11 @@ def _load_last_function_response(in_context_messages: list[Message]):
except (json.JSONDecodeError, KeyError):
raise ValueError(f"Invalid JSON format in message: {text_content}")
return None
def _maybe_get_approval_messages(messages: list[Message]) -> Tuple[Message | None, Message | None]:
if len(messages) >= 2:
maybe_approval_request, maybe_approval_response = messages[-2], messages[-1]
if maybe_approval_request.role == "approval" and maybe_approval_response.role == "approval":
return maybe_approval_request, maybe_approval_response
return None, None

10
letta/agents/letta_agent_v2.py
View File

@@ -13,6 +13,7 @@ from letta.agents.ephemeral_summary_agent import EphemeralSummaryAgent
from letta.agents.helpers import (
_build_rule_violation_result,
_load_last_function_response,
_maybe_get_approval_messages,
_pop_heartbeat,
_prepare_in_context_messages_no_persist_async,
_safe_load_tool_call_str,
@@ -379,7 +380,7 @@ class LettaAgentV2(BaseAgentV2):
try:
self.last_function_response = _load_last_function_response(messages)
valid_tools = await self._get_valid_tools()
approval_request, approval_response = await self._maybe_get_approval_messages(messages)
approval_request, approval_response = _maybe_get_approval_messages(messages)
if approval_request and approval_response:
tool_call = approval_request.tool_calls[0]
reasoning_content = approval_request.content
@@ -601,13 +602,6 @@ class LettaAgentV2(BaseAgentV2):
self.last_function_response = None
self.response_messages = []
async def _maybe_get_approval_messages(self, messages: list[Message]) -> Tuple[Message | None, Message | None]:
if len(messages) >= 2:
maybe_approval_request, maybe_approval_response = messages[-2], messages[-1]
if maybe_approval_request.role == "approval" and maybe_approval_response.role == "approval":
return maybe_approval_request, maybe_approval_response
return None, None
@trace_method
async def _check_run_cancellation(self, run_id) -> bool:
try: