Anton
052591281d
- Complete agent system with planning and execution - Bridge API with FastAPI server - WhatsApp-style mobile dashboard - Enhanced web search with query improvement - Security guards and policy checks - Comprehensive documentation
100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""Template resolution for step payloads.
|
|
|
|
Supported patterns:
|
|
{{ last.output_text }} — output_text of the most recent step in history
|
|
{{ steps.step-1.output_text }} — output_text of a specific step by step_id
|
|
|
|
Resolution is called BEFORE executor.execute(step) in agent_loop.
|
|
Tools never see unresolved templates.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any, Optional
|
|
|
|
from core.state import AgentState
|
|
|
|
# Matches {{ expr }} with optional surrounding whitespace
|
|
_TMPL_RE = re.compile(r"\{\{\s*([\w.\-]+)\s*\}\}")
|
|
|
|
|
|
def resolve_payload(
|
|
payload: dict[str, Any], state: AgentState
|
|
) -> tuple[dict[str, Any], bool, str]:
|
|
"""Resolve all template expressions in a step payload.
|
|
|
|
Returns:
|
|
(resolved_payload, success, error_message)
|
|
On failure: ({}, False, "reason")
|
|
On success: (resolved_dict, True, "")
|
|
"""
|
|
resolved: dict[str, Any] = {}
|
|
for key, value in payload.items():
|
|
if isinstance(value, str) and _TMPL_RE.search(value):
|
|
new_val, ok, err = _resolve_string(value, state)
|
|
if not ok:
|
|
return {}, False, err
|
|
resolved[key] = new_val
|
|
else:
|
|
resolved[key] = value
|
|
return resolved, True, ""
|
|
|
|
|
|
def has_templates(payload: dict[str, Any]) -> bool:
|
|
"""Return True if any string value in payload contains a template."""
|
|
return any(
|
|
isinstance(v, str) and bool(_TMPL_RE.search(v))
|
|
for v in payload.values()
|
|
)
|
|
|
|
|
|
def _resolve_string(s: str, state: AgentState) -> tuple[str, bool, str]:
|
|
"""Replace all {{ expr }} occurrences in s. Returns (result, ok, error)."""
|
|
errors: list[str] = []
|
|
|
|
def replacer(m: re.Match) -> str:
|
|
expr = m.group(1).strip()
|
|
val, ok, err = _eval_expr(expr, state)
|
|
if not ok:
|
|
errors.append(err)
|
|
return ""
|
|
return val if val is not None else ""
|
|
|
|
result = _TMPL_RE.sub(replacer, s)
|
|
if errors:
|
|
return "", False, errors[0]
|
|
return result, True, ""
|
|
|
|
|
|
def _eval_expr(expr: str, state: AgentState) -> tuple[Optional[str], bool, str]:
|
|
"""Evaluate a single template expression against agent state."""
|
|
|
|
# {{ last.output_text }}
|
|
if expr == "last.output_text":
|
|
val = state.get_last_output()
|
|
if val is None:
|
|
return None, False, (
|
|
"Template '{{ last.output_text }}' could not be resolved: "
|
|
"history is empty or last step produced no output"
|
|
)
|
|
return val, True, ""
|
|
|
|
# {{ steps.<step_id>.output_text }}
|
|
if expr.startswith("steps.") and expr.endswith("..output_text"):
|
|
# "steps.step-1.output_text" → split on first and last dot
|
|
# Use rsplit to handle step-ids that might contain dots (unlikely but safe)
|
|
without_prefix = expr[len("steps."):] # "step-1.output_text"
|
|
if without_prefix.endswith(".output_text"):
|
|
step_id = without_prefix[: -len(".output_text")] # "step-1"
|
|
if not step_id:
|
|
return None, False, f"Template '{{{{ {expr} }}}}' has empty step_id"
|
|
val = state.get_step_output(step_id)
|
|
if val is None:
|
|
return None, False, (
|
|
f"Template '{{{{ {expr} }}}}' could not be resolved: "
|
|
f"step '{step_id}' not found in history or has no output"
|
|
)
|
|
return val, True, ""
|
|
|
|
return None, False, f"Unknown template expression: '{{{{ {expr} }}}}'"
|