Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 28 additions & 15 deletions src/google/adk/code_executors/code_execution_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,31 +145,44 @@ def extract_code_and_truncate_content(
first_text_part = copy.deepcopy(text_parts[0])
response_text = '\n'.join([p.text for p in text_parts])

# Find the first code block.
leading_delimiter_pattern = '|'.join(d[0] for d in code_block_delimiters)
trailing_delimiter_pattern = '|'.join(d[1] for d in code_block_delimiters)
pattern = re.compile(
(
rf'(?P<prefix>.*?)({leading_delimiter_pattern})(?P<code>.*?)({trailing_delimiter_pattern})(?P<suffix>.*?)$'
).encode(),
re.DOTALL,
)
pattern_match = pattern.search(response_text.encode())
if pattern_match is None:
# Find the first code block using simple string search (avoids ReDoS from
# the previous regex with multiple .*? groups — see issue #5992).
best_start = -1
best_end = -1
best_lead_len = 0
best_trail_len = 0

for lead, trail in code_block_delimiters:
start_idx = response_text.find(lead)
if start_idx == -1:
continue
code_start = start_idx + len(lead)
end_idx = response_text.find(trail, code_start)
if end_idx == -1:
continue
# Pick the earliest occurring code block.
if best_start == -1 or start_idx < best_start:
best_start = start_idx
best_end = end_idx
best_lead_len = len(lead)
best_trail_len = len(trail)

if best_start == -1:
return

code_str = pattern_match.group('code').decode()
code_str = response_text[best_start + best_lead_len : best_end]
if not code_str:
return

content.parts = []
if pattern_match.group('prefix'):
first_text_part.text = pattern_match.group('prefix').decode()
prefix_text = response_text[:best_start]
if prefix_text:
first_text_part.text = prefix_text
content.parts.append(first_text_part)
content.parts.append(
CodeExecutionUtils.build_executable_code_part(code_str)
)
return pattern_match.group('code').decode()
return code_str

@staticmethod
def build_executable_code_part(code: str) -> types.Part:
Expand Down