gmf_forge_ai_orchestration.agents
Agents — ReAct, Plan-Execute, Reflexion, Chain-of-Thought, and A2A.
Choosing the right agent
ReActAgent
Use when the task requires tool calls (search, lookup, API calls).
Interleaves reasoning (Thought) and acting (Action/Observation) in a loop.
The system prompt MUST include the Thought:/Action:/Action Input: format
contract — without it the agent degrades to a single-pass answer with no
tool calls. Supports a custom system_prompt.
ChainOfThoughtAgent
Use when the task requires structured reasoning but NO tool calls.
A single LLM call that reasons inside <thinking> tags before answering.
Simpler and cheaper than ReAct — good for classification, analysis, or
multi-step math. Supports a custom system_prompt.
PlanExecuteAgent
Use when the task has clearly separable sequential steps known upfront.
Phase 1 asks the LLM to decompose the task into a JSON step list.
Phase 2 executes each step sequentially; each step can call tools via
an inner Thought/Action loop. Good for structured workflows (e.g., search
→ summarise → write). Supports custom plan_prompt and execute_prompt.
ReflexionAgent
Wrapper — not a standalone agent. Adds a self-critique-and-retry loop
around any inner agent. Use when output quality matters more than latency
and a second-pass correction is acceptable.
A2AClient
Use to call a separately-deployed A2A-compliant agent service.
Sends tasks via JSON-RPC 2.0 tasks/send (A2A protocol).
Discovers agent capabilities via GET /.well-known/agent.json (Agent Card).
No local LLM or tools needed.
A2AAdapter
Use on the receiving side of an A2A service.
Handles the A2A protocol boilerplate for discovery and tasks/send
dispatch while leaving business logic to the host agent.
1"""Agents — ReAct, Plan-Execute, Reflexion, Chain-of-Thought, and A2A. 2 3Choosing the right agent 4------------------------ 5 6:class:`ReActAgent` 7 **Use when the task requires tool calls** (search, lookup, API calls). 8 Interleaves reasoning (Thought) and acting (Action/Observation) in a loop. 9 The system prompt MUST include the ``Thought:/Action:/Action Input:`` format 10 contract — without it the agent degrades to a single-pass answer with no 11 tool calls. Supports a custom ``system_prompt``. 12 13:class:`ChainOfThoughtAgent` 14 **Use when the task requires structured reasoning but NO tool calls.** 15 A single LLM call that reasons inside ``<thinking>`` tags before answering. 16 Simpler and cheaper than ReAct — good for classification, analysis, or 17 multi-step math. Supports a custom ``system_prompt``. 18 19:class:`PlanExecuteAgent` 20 **Use when the task has clearly separable sequential steps known upfront.** 21 Phase 1 asks the LLM to decompose the task into a JSON step list. 22 Phase 2 executes each step sequentially; each step can call tools via 23 an inner Thought/Action loop. Good for structured workflows (e.g., search 24 → summarise → write). Supports custom ``plan_prompt`` and ``execute_prompt``. 25 26:class:`ReflexionAgent` 27 **Wrapper — not a standalone agent.** Adds a self-critique-and-retry loop 28 around any inner agent. Use when output quality matters more than latency 29 and a second-pass correction is acceptable. 30 31:class:`A2AClient` 32 **Use to call a separately-deployed A2A-compliant agent service.** 33 Sends tasks via JSON-RPC 2.0 ``tasks/send`` (A2A protocol). 34 Discovers agent capabilities via ``GET /.well-known/agent.json`` (Agent Card). 35 No local LLM or tools needed. 36 37:class:`A2AAdapter` 38 **Use on the receiving side of an A2A service.** 39 Handles the A2A protocol boilerplate for discovery and ``tasks/send`` 40 dispatch while leaving business logic to the host agent. 41""" 42 43from gmf_forge_ai_orchestration.agents.base import AgentResult, AgentStep, BaseAgent, TokenUsage 44from gmf_forge_ai_orchestration.agents.react_agent import ReActAgent 45from gmf_forge_ai_orchestration.agents.plan_execute_agent import PlanExecuteAgent 46from gmf_forge_ai_orchestration.agents.reflexion_agent import ReflexionAgent 47from gmf_forge_ai_orchestration.agents.chain_of_thought_agent import ChainOfThoughtAgent 48from gmf_forge_ai_orchestration.protocols.a2a.a2a_client import ( 49 A2AClient, 50 A2AClientError, 51) 52from gmf_forge_ai_orchestration.protocols.a2a.a2a_adapter import ( 53 A2AAdapter, 54 A2AAdapterError, 55) 56 57__all__ = [ 58 "AgentResult", 59 "AgentStep", 60 "BaseAgent", 61 "TokenUsage", 62 "ReActAgent", 63 "PlanExecuteAgent", 64 "ReflexionAgent", 65 "ChainOfThoughtAgent", 66 "A2AClient", 67 "A2AClientError", 68 "A2AAdapter", 69 "A2AAdapterError", 70]
@dataclass
class
AgentResult:
60@dataclass 61class AgentResult: 62 """The final result of an agent execution.""" 63 64 output: str 65 steps: List[AgentStep] = field(default_factory=list) 66 metadata: Dict[str, Any] = field(default_factory=dict) 67 success: bool = True 68 error: Optional[str] = None 69 token_usage: Optional["TokenUsage"] = None
The final result of an agent execution.
AgentResult( output: str, steps: List[AgentStep] = <factory>, metadata: Dict[str, Any] = <factory>, success: bool = True, error: Optional[str] = None, token_usage: Optional[TokenUsage] = None)
steps: List[AgentStep]
@dataclass
class
AgentStep:
23@dataclass 24class AgentStep: 25 """One thought/action/observation cycle within an agent execution.""" 26 27 thought: str 28 action: str 29 action_input: Dict[str, Any] = field(default_factory=dict) 30 observation: str = "" 31 metadata: Dict[str, Any] = field(default_factory=dict)
One thought/action/observation cycle within an agent execution.
class
BaseAgent(abc.ABC):
72class BaseAgent(ABC): 73 """ 74 Abstract base class for all agents. 75 76 Wires together: 77 - :class:`UnifiedLLMGateway` for LLM calls 78 - :class:`ToolRegistry` for tool discovery and execution 79 - :class:`BaseBehavior` pipeline applied around every execution 80 - :class:`BaseStateStore` for persisting conversation/step state 81 - Full shared-core observability stack (Logger, Metrics, PerformanceMonitor, Tracing) 82 83 Args: 84 llm_gateway: Required. The LLM gateway to use for completions. 85 tool_registry: Optional tool registry. Provides available tools to the agent. 86 behaviors: Ordered list of behaviors applied around each execution. 87 state_store: Optional state store for persisting steps and conversation. 88 agent_id: Stable identifier used in logging and metrics. Defaults to class name. 89 logger: Optional :class:`BasicLogger`. Created automatically if omitted. 90 metrics: Optional :class:`BasicMetricsCollector`. 91 performance_monitor: Optional :class:`BasicPerformanceMonitor`. 92 tracer: Optional :class:`TracingProvider`. Falls back to ``get_tracer()``. 93 """ 94 95 def __init__( 96 self, 97 llm_gateway: Optional["UnifiedLLMGateway"] = None, 98 tool_registry: Optional["ToolRegistry"] = None, 99 behaviors: Optional[List["BaseBehavior"]] = None, 100 state_store: Optional["BaseStateStore"] = None, 101 checkpoint_manager: Optional["CheckpointManager"] = None, 102 agent_id: Optional[str] = None, 103 logger: Optional[BasicLogger] = None, 104 metrics: Optional[BasicMetricsCollector] = None, 105 performance_monitor: Optional[BasicPerformanceMonitor] = None, 106 tracer: Optional[TracingProvider] = None, 107 ) -> None: 108 self.llm_gateway = llm_gateway 109 self.tool_registry = tool_registry 110 self.behaviors: List["BaseBehavior"] = behaviors or [] 111 self.state_store = state_store 112 self.checkpoint_manager = checkpoint_manager 113 self.agent_id = agent_id or self.__class__.__name__ 114 self._logger = logger or BasicLogger(f"gmf_forge_ai.agent.{self.agent_id}") 115 self._metrics = metrics 116 self._performance_monitor = performance_monitor 117 self._tracer = tracer or get_tracer() 118 119 # ------------------------------------------------------------------ 120 # Abstract interface 121 # ------------------------------------------------------------------ 122 123 @abstractmethod 124 async def execute( 125 self, task: str, context: Optional[Dict[str, Any]] = None 126 ) -> AgentResult: 127 """Execute the task and return a result.""" 128 129 @abstractmethod 130 async def stream_execute( 131 self, task: str, context: Optional[Dict[str, Any]] = None 132 ) -> AsyncIterator[AgentStep]: 133 """Execute the task, yielding each step as it completes.""" 134 135 # ------------------------------------------------------------------ 136 # Behavior pipeline helpers (called by subclasses) 137 # ------------------------------------------------------------------ 138 139 async def _apply_behaviors_before( 140 self, context: "BehaviorContext" 141 ) -> "BehaviorContext": 142 for behavior in self.behaviors: 143 context = await behavior.before_execute(context) 144 return context 145 146 async def _apply_behaviors_after( 147 self, context: "BehaviorContext", result: AgentResult 148 ) -> AgentResult: 149 for behavior in self.behaviors: 150 result = await behavior.after_execute(context, result) 151 return result 152 153 async def _apply_behaviors_on_error( 154 self, context: "BehaviorContext", error: Exception 155 ) -> Optional[AgentResult]: 156 for behavior in self.behaviors: 157 fallback = await behavior.on_error(context, error) 158 if fallback is not None: 159 return fallback 160 return None 161 162 # ------------------------------------------------------------------ 163 # Shared observability helpers (called by subclasses) 164 # ------------------------------------------------------------------ 165 166 def _log_execution_start(self, task: str) -> None: 167 self._logger.info( 168 "Agent execution started", agent_id=self.agent_id, task=task 169 ) 170 if self._metrics: 171 self._metrics.increment("agent.executions", agent_id=self.agent_id) 172 173 def _log_execution_end(self, task: str, success: bool, steps: int) -> None: 174 self._logger.info( 175 "Agent execution finished", 176 agent_id=self.agent_id, 177 task=task, 178 success=success, 179 steps=steps, 180 ) 181 if self._metrics: 182 self._metrics.histogram("agent.steps", steps, agent_id=self.agent_id) 183 184 def _log_execution_error(self, task: str, error: Exception) -> None: 185 self._logger.error( 186 "Agent execution error", 187 agent_id=self.agent_id, 188 task=task, 189 error=str(error), 190 error_type=type(error).__name__, 191 ) 192 if self._metrics: 193 self._metrics.increment("agent.errors", agent_id=self.agent_id)
Abstract base class for all agents.
Wires together:
UnifiedLLMGatewayfor LLM callsToolRegistryfor tool discovery and executionBaseBehaviorpipeline applied around every executionBaseStateStorefor persisting conversation/step state- Full shared-core observability stack (Logger, Metrics, PerformanceMonitor, Tracing)
Args:
llm_gateway: Required. The LLM gateway to use for completions.
tool_registry: Optional tool registry. Provides available tools to the agent.
behaviors: Ordered list of behaviors applied around each execution.
state_store: Optional state store for persisting steps and conversation.
agent_id: Stable identifier used in logging and metrics. Defaults to class name.
logger: Optional BasicLogger. Created automatically if omitted.
metrics: Optional BasicMetricsCollector.
performance_monitor: Optional BasicPerformanceMonitor.
tracer: Optional TracingProvider. Falls back to get_tracer().
behaviors: List[gmf_forge_ai_orchestration.BaseBehavior]
@abstractmethod
async def
execute( self, task: str, context: Optional[Dict[str, Any]] = None) -> AgentResult:
123 @abstractmethod 124 async def execute( 125 self, task: str, context: Optional[Dict[str, Any]] = None 126 ) -> AgentResult: 127 """Execute the task and return a result."""
Execute the task and return a result.
@abstractmethod
async def
stream_execute( self, task: str, context: Optional[Dict[str, Any]] = None) -> AsyncIterator[AgentStep]:
129 @abstractmethod 130 async def stream_execute( 131 self, task: str, context: Optional[Dict[str, Any]] = None 132 ) -> AsyncIterator[AgentStep]: 133 """Execute the task, yielding each step as it completes."""
Execute the task, yielding each step as it completes.
@dataclass
class
TokenUsage:
34@dataclass 35class TokenUsage: 36 """Token consumption recorded for a single agent execution. 37 38 Populated by agents that have access to LLM response usage data 39 (e.g. via BasicPerformanceMonitor). All fields default to zero so 40 the dataclass is safe to construct without token data when not available. 41 """ 42 43 input_tokens: int = 0 44 output_tokens: int = 0 45 total_tokens: int = 0 46 llm_calls: int = 0 47 model: str = "" 48 49 def add(self, other: "TokenUsage") -> "TokenUsage": 50 """Return a new TokenUsage that is the sum of self and other.""" 51 return TokenUsage( 52 input_tokens=self.input_tokens + other.input_tokens, 53 output_tokens=self.output_tokens + other.output_tokens, 54 total_tokens=self.total_tokens + other.total_tokens, 55 llm_calls=self.llm_calls + other.llm_calls, 56 model=self.model or other.model, 57 )
Token consumption recorded for a single agent execution.
Populated by agents that have access to LLM response usage data (e.g. via BasicPerformanceMonitor). All fields default to zero so the dataclass is safe to construct without token data when not available.
TokenUsage( input_tokens: int = 0, output_tokens: int = 0, total_tokens: int = 0, llm_calls: int = 0, model: str = '')
49 def add(self, other: "TokenUsage") -> "TokenUsage": 50 """Return a new TokenUsage that is the sum of self and other.""" 51 return TokenUsage( 52 input_tokens=self.input_tokens + other.input_tokens, 53 output_tokens=self.output_tokens + other.output_tokens, 54 total_tokens=self.total_tokens + other.total_tokens, 55 llm_calls=self.llm_calls + other.llm_calls, 56 model=self.model or other.model, 57 )
Return a new TokenUsage that is the sum of self and other.
56class ReActAgent(BaseAgent): 57 """ 58 ReAct agent: interleaves Reasoning (Thought) and Acting (Action/Observation). 59 60 **When to use:** Any task that requires tool calls (search, API lookup, 61 database queries). The LLM reasons step by step and acts on each step. 62 63 **When NOT to use:** Pure reasoning/analysis tasks with no tools — use 64 :class:`ChainOfThoughtAgent` instead (cheaper: one LLM call vs. many). 65 66 On each step the LLM produces a Thought, an Action (tool name or 67 ``"Final Answer"``), and an Action Input. If the action is a tool call, 68 the tool's output is fed back as an Observation and the loop continues. 69 70 .. warning:: 71 The system prompt MUST instruct the LLM to respond using the 72 ``Thought:/Action:/Action Input:`` format. Without this contract the 73 regex parser will not match, the agent will treat the first response 74 as a Final Answer, and **no tool calls will ever be made**. 75 Always include ``{tool_descriptions}`` in a custom prompt so the LLM 76 knows what tools are available. 77 78 Args: 79 max_steps: Maximum thought/action cycles before stopping (default: 10). 80 model: LLM model name passed to the gateway (optional). 81 temperature: Sampling temperature (default: 0.0 for determinism). 82 system_prompt: Override the default ReAct system prompt. Use 83 ``{tool_descriptions}`` as a placeholder if you want the agent's 84 available tools listed in your prompt. The task is always appended 85 separately and does not need a placeholder here. 86 87 All other args inherited from :class:`BaseAgent`. 88 """ 89 90 def __init__(self, *args: Any, max_steps: int = 10, model: Optional[str] = None, 91 temperature: float = 0.0, system_prompt: Optional[str] = None, 92 **kwargs: Any) -> None: 93 super().__init__(*args, **kwargs) 94 self.max_steps = max_steps 95 self.model = model 96 self.temperature = temperature 97 self._system_prompt = system_prompt 98 99 #: Default ReAct system prompt used when no ``system_prompt`` is passed. 100 #: Inspect this to understand the expected format or use it as a base 101 #: for your own customisations. 102 DEFAULT_SYSTEM_PROMPT: str = _REACT_SYSTEM 103 104 # ------------------------------------------------------------------ 105 # Internal prompt helpers 106 # ------------------------------------------------------------------ 107 108 def _tool_descriptions(self) -> str: 109 if not self.tool_registry: 110 return "No tools available." 111 tools = self.tool_registry.list_tools() 112 if not tools: 113 return "No tools available." 114 lines = [] 115 for t in tools: 116 lines.append(f"- {t.name}: {t.description}") 117 return "\n".join(lines) 118 119 def _build_prompt(self, task: str, history: List[AgentStep]) -> str: 120 template = self._system_prompt if self._system_prompt is not None else _REACT_SYSTEM 121 if "{tool_descriptions}" in template: 122 system = template.format(tool_descriptions=self._tool_descriptions()) 123 else: 124 system = template 125 turns = [f"Task: {task}\n"] 126 for step in history: 127 turns.append(f"Thought: {step.thought}") 128 turns.append(f"Action: {step.action}") 129 turns.append(f"Action Input: {json.dumps(step.action_input)}") 130 if step.observation: 131 turns.append(f"Observation: {step.observation}") 132 return system + "\n" + "\n".join(turns) 133 134 def _parse_step(self, text: str) -> Optional[AgentStep]: 135 def _coerce_action_input(raw_input: str) -> Dict[str, Any]: 136 try: 137 action_input = json.loads(raw_input) 138 if isinstance(action_input, dict): 139 return action_input 140 except (json.JSONDecodeError, ValueError): 141 pass 142 return {"raw": raw_input} 143 144 match = _REACT_STEP_PATTERN.search(text) 145 if match: 146 thought = match.group("thought").strip() 147 action = match.group("action").strip() 148 raw_input = match.group("action_input").strip() 149 action_input = _coerce_action_input(raw_input) 150 return AgentStep(thought=thought, action=action, action_input=action_input) 151 152 # Some models omit "Thought:" and return only Action/Action Input. 153 final_only = _REACT_FINAL_ONLY_PATTERN.search(text) 154 if final_only: 155 raw_input = final_only.group("action_input").strip() 156 action_input = _coerce_action_input(raw_input) 157 return AgentStep(thought="", action="Final Answer", action_input=action_input) 158 159 # Some models omit "Thought:" but still emit a tool call (not Final Answer). 160 # Execute the tool rather than treating the raw output as a Final Answer. 161 no_thought = _REACT_NO_THOUGHT_PATTERN.search(text) 162 if no_thought: 163 action = no_thought.group("action").strip() 164 raw_input = no_thought.group("action_input").strip() 165 action_input = _coerce_action_input(raw_input) 166 return AgentStep(thought="", action=action, action_input=action_input) 167 168 return None 169 170 def _extract_final_answer(self, step: AgentStep) -> str: 171 """Return normalized final answer text from a Final Answer step.""" 172 answer_obj: Any 173 if isinstance(step.action_input, dict) and "answer" in step.action_input: 174 answer_obj = step.action_input["answer"] 175 else: 176 answer_obj = step.action_input 177 178 answer = str(answer_obj).strip() 179 180 # Unwrap nested ReAct control text accidentally returned as answer body. 181 for _ in range(3): 182 nested = _REACT_FINAL_ONLY_PATTERN.search(answer) 183 if nested: 184 raw_input = nested.group("action_input").strip() 185 elif answer.startswith("Final Answer:"): 186 raw_input = answer[len("Final Answer:"):].strip() 187 else: 188 break 189 try: 190 parsed = json.loads(raw_input) 191 if isinstance(parsed, dict) and "answer" in parsed: 192 answer = str(parsed["answer"]).strip() 193 else: 194 answer = raw_input 195 except (json.JSONDecodeError, ValueError): 196 answer = raw_input 197 198 return answer.strip() 199 200 async def _save_checkpoint( 201 self, 202 execution_id: str, 203 task: str, 204 context: Dict[str, Any], 205 steps: List[AgentStep], 206 step_number: int, 207 ) -> None: 208 if not self.checkpoint_manager: 209 return 210 await self.checkpoint_manager.save( 211 agent_id=self.agent_id, 212 execution_id=execution_id, 213 state={ 214 "execution_id": execution_id, 215 "task": task, 216 "context": context, 217 "step_number": step_number, 218 "steps": [dataclasses.asdict(s) for s in steps], 219 }, 220 metadata={"step_number": step_number}, 221 ) 222 223 async def resume_from( 224 self, execution_id: str, from_step: Optional[int] = None 225 ) -> AgentResult: 226 """Resume a ReAct execution from its latest (or selected) checkpoint.""" 227 if not self.checkpoint_manager: 228 raise ValueError("Checkpoint manager is not configured for this agent") 229 230 checkpoint = None 231 if from_step is not None: 232 checkpoints = await self.checkpoint_manager.list_by_execution(execution_id) 233 for ckpt in checkpoints: 234 if ckpt.state.get("step_number") == from_step: 235 checkpoint = ckpt 236 break 237 else: 238 checkpoint = await self.checkpoint_manager.load_latest_for_execution(execution_id) 239 240 if checkpoint is None: 241 raise ValueError(f"No checkpoint found for execution_id={execution_id}") 242 243 state = checkpoint.state 244 task = str(state.get("task", "")) 245 context = state.get("context") or {} 246 raw_steps = state.get("steps") or [] 247 steps = [AgentStep(**raw) for raw in raw_steps if isinstance(raw, dict)] 248 249 # If the checkpoint already contains a final answer, return idempotently. 250 if steps and steps[-1].action == "Final Answer": 251 answer = self._extract_final_answer(steps[-1]) 252 return AgentResult( 253 output=str(answer), 254 steps=steps, 255 success=True, 256 metadata={"execution_id": execution_id}, 257 ) 258 259 ctx = BehaviorContext( 260 agent_id=self.agent_id, 261 task=task, 262 execution_id=execution_id, 263 last_completed_step=int(state.get("step_number", len(steps) - 1)), 264 metadata={"resume_agent": self}, 265 ) 266 ctx = await self._apply_behaviors_before(ctx) 267 self._log_execution_start(task) 268 269 for step_num in range(len(steps), self.max_steps): 270 prompt = self._build_prompt(task, steps) 271 response = await self.llm_gateway.complete( 272 prompt, model=self.model, temperature=self.temperature 273 ) 274 step = self._parse_step(response.content) 275 if step is None: 276 step = AgentStep( 277 thought="Could not parse structured response.", 278 action="Final Answer", 279 action_input={"answer": self._extract_final_answer(AgentStep(thought="", action="Final Answer", action_input={"answer": response.content}))}, 280 ) 281 282 if step.action == "Final Answer": 283 answer = self._extract_final_answer(step) 284 step.observation = answer 285 steps.append(step) 286 await self._save_checkpoint(execution_id, task, context, steps, step_num) 287 ctx.last_completed_step = step_num 288 result = AgentResult( 289 output=answer, 290 steps=steps, 291 success=True, 292 metadata={"execution_id": execution_id}, 293 ) 294 result = await self._apply_behaviors_after(ctx, result) 295 self._log_execution_end(task, success=True, steps=len(steps)) 296 return result 297 298 observation = "" 299 if self.tool_registry: 300 try: 301 tool_result = await self.tool_registry.execute(step.action, **step.action_input) 302 observation = str(tool_result) 303 except Exception as tool_exc: 304 observation = f"Tool error: {tool_exc}" 305 else: 306 observation = f"No tool registry — cannot execute '{step.action}'" 307 308 step.observation = observation 309 steps.append(step) 310 await self._save_checkpoint(execution_id, task, context, steps, step_num) 311 ctx.last_completed_step = step_num 312 313 final_output = steps[-1].observation if steps else "No output" 314 result = AgentResult( 315 output=final_output, 316 steps=steps, 317 success=False, 318 error=f"Reached max_steps={self.max_steps} without Final Answer", 319 metadata={"execution_id": execution_id}, 320 ) 321 result = await self._apply_behaviors_after(ctx, result) 322 self._log_execution_end(task, success=False, steps=len(steps)) 323 return result 324 325 # ------------------------------------------------------------------ 326 # BaseAgent implementation 327 # ------------------------------------------------------------------ 328 329 async def execute( 330 self, task: str, context: Optional[Dict[str, Any]] = None 331 ) -> AgentResult: 332 execution_id = str(uuid4()) 333 run_context = context or {} 334 ctx = BehaviorContext( 335 agent_id=self.agent_id, 336 task=task, 337 execution_id=execution_id, 338 metadata={"resume_agent": self, **run_context}, 339 ) 340 ctx = await self._apply_behaviors_before(ctx) 341 self._log_execution_start(task) 342 343 while True: 344 _retry = False 345 _control_exit = None # HumanApprovalRequired or PendingApproval — not errors 346 steps: List[AgentStep] = [] 347 348 with self._tracer.trace( 349 "react_agent.execute", input=task, metadata={"agent_id": self.agent_id} 350 ) as trace: 351 try: 352 for _ in range(self.max_steps): 353 prompt = self._build_prompt(task, steps) 354 355 with trace.generation( 356 "llm_call", model=self.model or "default", input=prompt 357 ) as gen: 358 perf_id = None 359 if self._performance_monitor: 360 perf_id = self._performance_monitor.start_request( 361 provider="llm_gateway", model=self.model or "default" 362 ) 363 try: 364 response = await self.llm_gateway.complete( 365 prompt, model=self.model, temperature=self.temperature 366 ) 367 if self._performance_monitor and perf_id: 368 self._performance_monitor.end_request( 369 request_id=perf_id, 370 prompt_tokens=response.usage.get("prompt_tokens", 0), 371 completion_tokens=response.usage.get("completion_tokens", 0), 372 success=True, 373 ) 374 gen.set_output(response.content) 375 gen.set_token_usage(**response.usage) 376 except Exception as exc: 377 if self._performance_monitor and perf_id: 378 self._performance_monitor.end_request( 379 request_id=perf_id, 380 prompt_tokens=0, 381 completion_tokens=0, 382 success=False, 383 error=str(exc), 384 ) 385 raise 386 387 step = self._parse_step(response.content) 388 if step is None: 389 # Unparseable response — treat as final answer 390 step = AgentStep( 391 thought="Could not parse structured response.", 392 action="Final Answer", 393 action_input={"answer": self._extract_final_answer(AgentStep(thought="", action="Final Answer", action_input={"answer": response.content}))}, 394 ) 395 396 with trace.span("step", input=step.action) as span: 397 if step.action == "Final Answer": 398 answer = self._extract_final_answer(step) 399 step.observation = answer 400 steps.append(step) 401 span.set_output(answer) 402 403 # Persist to state store if available 404 if self.state_store: 405 await self.state_store.set( 406 f"agent:{self.agent_id}:last_steps", 407 [vars(s) for s in steps], 408 ttl=self.checkpoint_manager.default_ttl if self.checkpoint_manager else None, 409 ) 410 411 await self._save_checkpoint( 412 execution_id, 413 task, 414 run_context, 415 steps, 416 len(steps) - 1, 417 ) 418 ctx.last_completed_step = len(steps) - 1 419 420 result = AgentResult( 421 output=answer, 422 steps=steps, 423 success=True, 424 metadata={"execution_id": execution_id}, 425 ) 426 try: 427 result = await self._apply_behaviors_after(ctx, result) 428 except (HumanApprovalRequired, PendingApproval) as ctrl_exc: 429 _control_exit = ctrl_exc 430 431 if _control_exit is None: 432 self._log_execution_end(task, success=True, steps=len(steps)) 433 trace.set_output(answer) 434 return result 435 else: 436 # Control-flow exit (HumanApprovalRequired / PendingApproval) 437 # span output already set above; also set trace output so 438 # it logs the answer rather than finishing with no output. 439 trace.set_output(answer) 440 441 else: 442 # Tool call 443 observation = "" 444 if self.tool_registry: 445 try: 446 tool_result = await self.tool_registry.execute( 447 step.action, **step.action_input 448 ) 449 observation = str(tool_result) 450 except Exception as tool_exc: 451 observation = f"Tool error: {tool_exc}" 452 else: 453 observation = f"No tool registry — cannot execute '{step.action}'" 454 455 step.observation = observation 456 span.set_output(observation) 457 steps.append(step) 458 await self._save_checkpoint( 459 execution_id, 460 task, 461 run_context, 462 steps, 463 len(steps) - 1, 464 ) 465 ctx.last_completed_step = len(steps) - 1 466 467 if _control_exit is not None: 468 break # exit for loop; span already closed cleanly 469 470 # Max steps reached (or for loop broken by control-flow signal) 471 if _control_exit is None: 472 final_output = steps[-1].observation if steps else "No output" 473 result = AgentResult( 474 output=final_output, 475 steps=steps, 476 success=False, 477 error=f"Reached max_steps={self.max_steps} without Final Answer", 478 metadata={"execution_id": execution_id}, 479 ) 480 result = await self._apply_behaviors_after(ctx, result) 481 self._log_execution_end(task, success=False, steps=len(steps)) 482 trace.set_output(final_output) 483 return result 484 # else: control-flow exit — trace closes cleanly, re-raised below 485 486 except Exception as exc: 487 self._log_execution_error(task, exc) 488 fallback = await self._apply_behaviors_on_error(ctx, exc) 489 if fallback is RETRY_SENTINEL: 490 # RetryBehavior slept and wants us to retry from scratch. 491 # Mark the trace as failed so it logs "failed" not "finished". 492 trace.set_error(exc) 493 ctx.attempt += 1 494 _retry = True 495 elif fallback is not None: 496 return fallback 497 else: 498 raise 499 500 if _control_exit is not None: 501 raise _control_exit 502 503 if not _retry: 504 break 505 506 async def stream_execute( 507 self, task: str, context: Optional[Dict[str, Any]] = None 508 ) -> AsyncIterator[AgentStep]: 509 ctx = BehaviorContext(agent_id=self.agent_id, task=task) 510 ctx = await self._apply_behaviors_before(ctx) 511 self._log_execution_start(task) 512 513 steps: List[AgentStep] = [] 514 for _ in range(self.max_steps): 515 prompt = self._build_prompt(task, steps) 516 response = await self.llm_gateway.complete( 517 prompt, model=self.model, temperature=self.temperature 518 ) 519 step = self._parse_step(response.content) 520 if step is None: 521 step = AgentStep( 522 thought="Unparseable response.", 523 action="Final Answer", 524 action_input={"answer": response.content}, 525 ) 526 527 if step.action == "Final Answer": 528 step.observation = self._extract_final_answer(step) 529 steps.append(step) 530 yield step 531 return 532 533 if self.tool_registry: 534 try: 535 tool_result = await self.tool_registry.execute(step.action, **step.action_input) 536 step.observation = str(tool_result) 537 except Exception as tool_exc: 538 step.observation = f"Tool error: {tool_exc}" 539 else: 540 step.observation = f"No tool registry — cannot execute '{step.action}'" 541 542 steps.append(step) 543 yield step
ReAct agent: interleaves Reasoning (Thought) and Acting (Action/Observation).
When to use: Any task that requires tool calls (search, API lookup, database queries). The LLM reasons step by step and acts on each step.
When NOT to use: Pure reasoning/analysis tasks with no tools — use
ChainOfThoughtAgent instead (cheaper: one LLM call vs. many).
On each step the LLM produces a Thought, an Action (tool name or
"Final Answer"), and an Action Input. If the action is a tool call,
the tool's output is fed back as an Observation and the loop continues.
The system prompt MUST instruct the LLM to respond using the
Thought:/Action:/Action Input: format. Without this contract the
regex parser will not match, the agent will treat the first response
as a Final Answer, and no tool calls will ever be made.
Always include {tool_descriptions} in a custom prompt so the LLM
knows what tools are available.
Args:
max_steps: Maximum thought/action cycles before stopping (default: 10).
model: LLM model name passed to the gateway (optional).
temperature: Sampling temperature (default: 0.0 for determinism).
system_prompt: Override the default ReAct system prompt. Use
{tool_descriptions} as a placeholder if you want the agent's
available tools listed in your prompt. The task is always appended
separately and does not need a placeholder here.
All other args inherited from BaseAgent.
ReActAgent( *args: Any, max_steps: int = 10, model: Optional[str] = None, temperature: float = 0.0, system_prompt: Optional[str] = None, **kwargs: Any)
90 def __init__(self, *args: Any, max_steps: int = 10, model: Optional[str] = None, 91 temperature: float = 0.0, system_prompt: Optional[str] = None, 92 **kwargs: Any) -> None: 93 super().__init__(*args, **kwargs) 94 self.max_steps = max_steps 95 self.model = model 96 self.temperature = temperature 97 self._system_prompt = system_prompt
DEFAULT_SYSTEM_PROMPT: str =
'You are an AI assistant that reasons step by step before acting.\n\nYou have access to the following tools:\n{tool_descriptions}\n\nRespond using EXACTLY this format for each step:\nThought: <your reasoning>\nAction: <tool name or "Final Answer">\nAction Input: <JSON object with tool arguments, or your final answer string>\n\nWhen you have enough information, use:\nAction: Final Answer\nAction Input: {{"answer": "<your complete answer>"}}\n'
223 async def resume_from( 224 self, execution_id: str, from_step: Optional[int] = None 225 ) -> AgentResult: 226 """Resume a ReAct execution from its latest (or selected) checkpoint.""" 227 if not self.checkpoint_manager: 228 raise ValueError("Checkpoint manager is not configured for this agent") 229 230 checkpoint = None 231 if from_step is not None: 232 checkpoints = await self.checkpoint_manager.list_by_execution(execution_id) 233 for ckpt in checkpoints: 234 if ckpt.state.get("step_number") == from_step: 235 checkpoint = ckpt 236 break 237 else: 238 checkpoint = await self.checkpoint_manager.load_latest_for_execution(execution_id) 239 240 if checkpoint is None: 241 raise ValueError(f"No checkpoint found for execution_id={execution_id}") 242 243 state = checkpoint.state 244 task = str(state.get("task", "")) 245 context = state.get("context") or {} 246 raw_steps = state.get("steps") or [] 247 steps = [AgentStep(**raw) for raw in raw_steps if isinstance(raw, dict)] 248 249 # If the checkpoint already contains a final answer, return idempotently. 250 if steps and steps[-1].action == "Final Answer": 251 answer = self._extract_final_answer(steps[-1]) 252 return AgentResult( 253 output=str(answer), 254 steps=steps, 255 success=True, 256 metadata={"execution_id": execution_id}, 257 ) 258 259 ctx = BehaviorContext( 260 agent_id=self.agent_id, 261 task=task, 262 execution_id=execution_id, 263 last_completed_step=int(state.get("step_number", len(steps) - 1)), 264 metadata={"resume_agent": self}, 265 ) 266 ctx = await self._apply_behaviors_before(ctx) 267 self._log_execution_start(task) 268 269 for step_num in range(len(steps), self.max_steps): 270 prompt = self._build_prompt(task, steps) 271 response = await self.llm_gateway.complete( 272 prompt, model=self.model, temperature=self.temperature 273 ) 274 step = self._parse_step(response.content) 275 if step is None: 276 step = AgentStep( 277 thought="Could not parse structured response.", 278 action="Final Answer", 279 action_input={"answer": self._extract_final_answer(AgentStep(thought="", action="Final Answer", action_input={"answer": response.content}))}, 280 ) 281 282 if step.action == "Final Answer": 283 answer = self._extract_final_answer(step) 284 step.observation = answer 285 steps.append(step) 286 await self._save_checkpoint(execution_id, task, context, steps, step_num) 287 ctx.last_completed_step = step_num 288 result = AgentResult( 289 output=answer, 290 steps=steps, 291 success=True, 292 metadata={"execution_id": execution_id}, 293 ) 294 result = await self._apply_behaviors_after(ctx, result) 295 self._log_execution_end(task, success=True, steps=len(steps)) 296 return result 297 298 observation = "" 299 if self.tool_registry: 300 try: 301 tool_result = await self.tool_registry.execute(step.action, **step.action_input) 302 observation = str(tool_result) 303 except Exception as tool_exc: 304 observation = f"Tool error: {tool_exc}" 305 else: 306 observation = f"No tool registry — cannot execute '{step.action}'" 307 308 step.observation = observation 309 steps.append(step) 310 await self._save_checkpoint(execution_id, task, context, steps, step_num) 311 ctx.last_completed_step = step_num 312 313 final_output = steps[-1].observation if steps else "No output" 314 result = AgentResult( 315 output=final_output, 316 steps=steps, 317 success=False, 318 error=f"Reached max_steps={self.max_steps} without Final Answer", 319 metadata={"execution_id": execution_id}, 320 ) 321 result = await self._apply_behaviors_after(ctx, result) 322 self._log_execution_end(task, success=False, steps=len(steps)) 323 return result
Resume a ReAct execution from its latest (or selected) checkpoint.
329 async def execute( 330 self, task: str, context: Optional[Dict[str, Any]] = None 331 ) -> AgentResult: 332 execution_id = str(uuid4()) 333 run_context = context or {} 334 ctx = BehaviorContext( 335 agent_id=self.agent_id, 336 task=task, 337 execution_id=execution_id, 338 metadata={"resume_agent": self, **run_context}, 339 ) 340 ctx = await self._apply_behaviors_before(ctx) 341 self._log_execution_start(task) 342 343 while True: 344 _retry = False 345 _control_exit = None # HumanApprovalRequired or PendingApproval — not errors 346 steps: List[AgentStep] = [] 347 348 with self._tracer.trace( 349 "react_agent.execute", input=task, metadata={"agent_id": self.agent_id} 350 ) as trace: 351 try: 352 for _ in range(self.max_steps): 353 prompt = self._build_prompt(task, steps) 354 355 with trace.generation( 356 "llm_call", model=self.model or "default", input=prompt 357 ) as gen: 358 perf_id = None 359 if self._performance_monitor: 360 perf_id = self._performance_monitor.start_request( 361 provider="llm_gateway", model=self.model or "default" 362 ) 363 try: 364 response = await self.llm_gateway.complete( 365 prompt, model=self.model, temperature=self.temperature 366 ) 367 if self._performance_monitor and perf_id: 368 self._performance_monitor.end_request( 369 request_id=perf_id, 370 prompt_tokens=response.usage.get("prompt_tokens", 0), 371 completion_tokens=response.usage.get("completion_tokens", 0), 372 success=True, 373 ) 374 gen.set_output(response.content) 375 gen.set_token_usage(**response.usage) 376 except Exception as exc: 377 if self._performance_monitor and perf_id: 378 self._performance_monitor.end_request( 379 request_id=perf_id, 380 prompt_tokens=0, 381 completion_tokens=0, 382 success=False, 383 error=str(exc), 384 ) 385 raise 386 387 step = self._parse_step(response.content) 388 if step is None: 389 # Unparseable response — treat as final answer 390 step = AgentStep( 391 thought="Could not parse structured response.", 392 action="Final Answer", 393 action_input={"answer": self._extract_final_answer(AgentStep(thought="", action="Final Answer", action_input={"answer": response.content}))}, 394 ) 395 396 with trace.span("step", input=step.action) as span: 397 if step.action == "Final Answer": 398 answer = self._extract_final_answer(step) 399 step.observation = answer 400 steps.append(step) 401 span.set_output(answer) 402 403 # Persist to state store if available 404 if self.state_store: 405 await self.state_store.set( 406 f"agent:{self.agent_id}:last_steps", 407 [vars(s) for s in steps], 408 ttl=self.checkpoint_manager.default_ttl if self.checkpoint_manager else None, 409 ) 410 411 await self._save_checkpoint( 412 execution_id, 413 task, 414 run_context, 415 steps, 416 len(steps) - 1, 417 ) 418 ctx.last_completed_step = len(steps) - 1 419 420 result = AgentResult( 421 output=answer, 422 steps=steps, 423 success=True, 424 metadata={"execution_id": execution_id}, 425 ) 426 try: 427 result = await self._apply_behaviors_after(ctx, result) 428 except (HumanApprovalRequired, PendingApproval) as ctrl_exc: 429 _control_exit = ctrl_exc 430 431 if _control_exit is None: 432 self._log_execution_end(task, success=True, steps=len(steps)) 433 trace.set_output(answer) 434 return result 435 else: 436 # Control-flow exit (HumanApprovalRequired / PendingApproval) 437 # span output already set above; also set trace output so 438 # it logs the answer rather than finishing with no output. 439 trace.set_output(answer) 440 441 else: 442 # Tool call 443 observation = "" 444 if self.tool_registry: 445 try: 446 tool_result = await self.tool_registry.execute( 447 step.action, **step.action_input 448 ) 449 observation = str(tool_result) 450 except Exception as tool_exc: 451 observation = f"Tool error: {tool_exc}" 452 else: 453 observation = f"No tool registry — cannot execute '{step.action}'" 454 455 step.observation = observation 456 span.set_output(observation) 457 steps.append(step) 458 await self._save_checkpoint( 459 execution_id, 460 task, 461 run_context, 462 steps, 463 len(steps) - 1, 464 ) 465 ctx.last_completed_step = len(steps) - 1 466 467 if _control_exit is not None: 468 break # exit for loop; span already closed cleanly 469 470 # Max steps reached (or for loop broken by control-flow signal) 471 if _control_exit is None: 472 final_output = steps[-1].observation if steps else "No output" 473 result = AgentResult( 474 output=final_output, 475 steps=steps, 476 success=False, 477 error=f"Reached max_steps={self.max_steps} without Final Answer", 478 metadata={"execution_id": execution_id}, 479 ) 480 result = await self._apply_behaviors_after(ctx, result) 481 self._log_execution_end(task, success=False, steps=len(steps)) 482 trace.set_output(final_output) 483 return result 484 # else: control-flow exit — trace closes cleanly, re-raised below 485 486 except Exception as exc: 487 self._log_execution_error(task, exc) 488 fallback = await self._apply_behaviors_on_error(ctx, exc) 489 if fallback is RETRY_SENTINEL: 490 # RetryBehavior slept and wants us to retry from scratch. 491 # Mark the trace as failed so it logs "failed" not "finished". 492 trace.set_error(exc) 493 ctx.attempt += 1 494 _retry = True 495 elif fallback is not None: 496 return fallback 497 else: 498 raise 499 500 if _control_exit is not None: 501 raise _control_exit 502 503 if not _retry: 504 break
Execute the task and return a result.
async def
stream_execute( self, task: str, context: Optional[Dict[str, Any]] = None) -> AsyncIterator[AgentStep]:
506 async def stream_execute( 507 self, task: str, context: Optional[Dict[str, Any]] = None 508 ) -> AsyncIterator[AgentStep]: 509 ctx = BehaviorContext(agent_id=self.agent_id, task=task) 510 ctx = await self._apply_behaviors_before(ctx) 511 self._log_execution_start(task) 512 513 steps: List[AgentStep] = [] 514 for _ in range(self.max_steps): 515 prompt = self._build_prompt(task, steps) 516 response = await self.llm_gateway.complete( 517 prompt, model=self.model, temperature=self.temperature 518 ) 519 step = self._parse_step(response.content) 520 if step is None: 521 step = AgentStep( 522 thought="Unparseable response.", 523 action="Final Answer", 524 action_input={"answer": response.content}, 525 ) 526 527 if step.action == "Final Answer": 528 step.observation = self._extract_final_answer(step) 529 steps.append(step) 530 yield step 531 return 532 533 if self.tool_registry: 534 try: 535 tool_result = await self.tool_registry.execute(step.action, **step.action_input) 536 step.observation = str(tool_result) 537 except Exception as tool_exc: 538 step.observation = f"Tool error: {tool_exc}" 539 else: 540 step.observation = f"No tool registry — cannot execute '{step.action}'" 541 542 steps.append(step) 543 yield step
Execute the task, yielding each step as it completes.
63class PlanExecuteAgent(BaseAgent): 64 """ 65 Two-phase agent: LLM plans all steps first, then executes each step sequentially. 66 67 Phase 1 — Plan: Ask the LLM to decompose the task into a JSON list of steps. 68 Phase 2 — Execute: Feed each step back to the LLM (with accumulated context) to 69 produce a result. Each step runs an inner tool-calling loop — if the LLM emits a 70 ``Thought/Action/Action Input`` block referencing a registered tool, the tool is 71 invoked and its observation is fed back for the LLM to produce a final step result. 72 73 Args: 74 max_plan_steps: Maximum number of planned steps allowed (default: 10). 75 max_tool_calls_per_step: Maximum tool calls allowed within a single plan step 76 (default: 3). 77 model: LLM model name (optional). 78 temperature: Sampling temperature (default: 0.1). 79 plan_prompt: Override the planning prompt. Must contain ``{task}``. 80 execute_prompt: Override the step-execution prompt. Must contain 81 ``{task}``, ``{plan}``, ``{previous_results}``, ``{step_num}``, 82 ``{current_step}``, and ``{tool_section}``. 83 84 All other args inherited from :class:`BaseAgent`. 85 """ 86 87 def __init__( 88 self, 89 *args: Any, 90 max_plan_steps: int = 10, 91 max_tool_calls_per_step: int = 3, 92 model: Optional[str] = None, 93 temperature: float = 0.1, 94 plan_prompt: Optional[str] = None, 95 execute_prompt: Optional[str] = None, 96 **kwargs: Any, 97 ) -> None: 98 super().__init__(*args, **kwargs) 99 self.max_plan_steps = max_plan_steps 100 self.max_tool_calls_per_step = max_tool_calls_per_step 101 self.model = model 102 self.temperature = temperature 103 self._plan_prompt = plan_prompt 104 self._execute_prompt = execute_prompt 105 106 #: Default planning prompt. Must contain ``{task}`` if overriding. 107 DEFAULT_PLAN_PROMPT: str = _PLAN_PROMPT 108 #: Default step-execution prompt. Must contain ``{task}``, ``{plan}``, 109 #: ``{previous_results}``, ``{step_num}``, ``{current_step}``, and 110 #: ``{tool_section}`` if overriding (``{tool_section}`` is auto-populated 111 #: from the tool registry; pass an empty string if no tools are needed). 112 DEFAULT_EXECUTE_PROMPT: str = _EXECUTE_PROMPT 113 114 # ------------------------------------------------------------------ 115 # Phase 1: Plan 116 # ------------------------------------------------------------------ 117 118 async def _plan(self, task: str) -> List[str]: 119 template = self._plan_prompt if self._plan_prompt is not None else _PLAN_PROMPT 120 prompt = template.format(task=task) 121 response = await self.llm_gateway.complete( 122 prompt, model=self.model, temperature=0.0 123 ) 124 raw = response.content.strip() 125 # Extract JSON array even if wrapped in markdown fences 126 match = re.search(r"\[.*\]", raw, re.DOTALL) 127 if match: 128 try: 129 steps = json.loads(match.group()) 130 if isinstance(steps, list): 131 return [str(s) for s in steps[: self.max_plan_steps]] 132 except (json.JSONDecodeError, ValueError): 133 pass 134 # Fallback: treat each line as a step 135 lines = [ln.strip().lstrip("0123456789.-) ") for ln in raw.splitlines() if ln.strip()] 136 return lines[: self.max_plan_steps] 137 138 # ------------------------------------------------------------------ 139 # Phase 2: Execute a single step (with optional tool calls) 140 # ------------------------------------------------------------------ 141 142 def _tool_descriptions(self) -> str: 143 if not self.tool_registry: 144 return "" 145 tools = self.tool_registry.list_tools() 146 if not tools: 147 return "" 148 return "\n".join(f"- {t.name}: {t.description}" for t in tools) 149 150 async def _execute_step( 151 self, task: str, plan: List[str], step_num: int, previous: List[str] 152 ) -> str: 153 tool_descriptions = self._tool_descriptions() 154 tool_section = ( 155 _TOOL_SECTION.format(tool_descriptions=tool_descriptions) 156 if tool_descriptions 157 else "" 158 ) 159 template = self._execute_prompt if self._execute_prompt is not None else _EXECUTE_PROMPT 160 prompt = template.format( 161 task=task, 162 plan="\n".join(f"{i+1}. {s}" for i, s in enumerate(plan)), 163 previous_results="\n".join( 164 f"Step {i+1} result: {r}" for i, r in enumerate(previous) 165 ) or "None yet.", 166 step_num=step_num, 167 current_step=plan[step_num - 1], 168 tool_section=tool_section, 169 ) 170 171 # Inner tool-calling loop: allow up to max_tool_calls tool calls per step 172 conversation = prompt 173 for _ in range(self.max_tool_calls_per_step): 174 response = await self.llm_gateway.complete( 175 conversation, model=self.model, temperature=self.temperature 176 ) 177 raw = response.content.strip() 178 179 # Check if the LLM wants to call a tool. Accept the canonical 180 # Thought/Action/Action Input block, and also tolerate outputs that 181 # omit the leading "Thought:" line (some models drop it). 182 match = _STEP_PATTERN.search(raw) or _STEP_NO_THOUGHT_PATTERN.search(raw) 183 if match and self.tool_registry: 184 action = match.group("action").strip() 185 raw_input = match.group("action_input").strip() 186 try: 187 action_input = json.loads(raw_input) 188 if not isinstance(action_input, dict): 189 action_input = {"raw": raw_input} 190 except (json.JSONDecodeError, ValueError): 191 action_input = {"raw": raw_input} 192 193 try: 194 tool_result = await self.tool_registry.execute(action, **action_input) 195 observation = str(tool_result) 196 except Exception as exc: 197 observation = f"Tool error: {exc}" 198 199 # Append the tool call + observation to the conversation and continue 200 conversation = ( 201 conversation 202 + f"\n{raw}\nObservation: {observation}\n" 203 + "Now provide the final result for this step based on the above observation." 204 ) 205 else: 206 # No tool call pattern — treat the response as the step result 207 return raw 208 209 # Exhausted tool call budget — return last response 210 return raw 211 212 async def _save_checkpoint( 213 self, 214 execution_id: str, 215 task: str, 216 context: Dict[str, Any], 217 plan: List[str], 218 steps: List[AgentStep], 219 step_results: List[str], 220 step_number: int, 221 ) -> None: 222 if not self.checkpoint_manager: 223 return 224 await self.checkpoint_manager.save( 225 agent_id=self.agent_id, 226 execution_id=execution_id, 227 state={ 228 "execution_id": execution_id, 229 "task": task, 230 "context": context, 231 "plan": plan, 232 "step_number": step_number, 233 "step_results": list(step_results), 234 "steps": [dataclasses.asdict(s) for s in steps], 235 }, 236 metadata={"step_number": step_number}, 237 ttl=86400, 238 ) 239 240 async def resume_from( 241 self, execution_id: str, from_step: Optional[int] = None 242 ) -> AgentResult: 243 """Resume a Plan-Execute run from latest (or selected) checkpoint.""" 244 if not self.checkpoint_manager: 245 raise ValueError("Checkpoint manager is not configured for this agent") 246 247 checkpoint = None 248 if from_step is not None: 249 checkpoints = await self.checkpoint_manager.list_by_execution(execution_id) 250 for ckpt in checkpoints: 251 if ckpt.state.get("step_number") == from_step: 252 checkpoint = ckpt 253 break 254 else: 255 checkpoint = await self.checkpoint_manager.load_latest_for_execution(execution_id) 256 257 if checkpoint is None: 258 raise ValueError(f"No checkpoint found for execution_id={execution_id}") 259 260 state = checkpoint.state 261 task = str(state.get("task", "")) 262 context = state.get("context") or {} 263 plan = [str(s) for s in state.get("plan") or []] 264 raw_steps = state.get("steps") or [] 265 steps = [AgentStep(**raw) for raw in raw_steps if isinstance(raw, dict)] 266 step_results = [str(r) for r in state.get("step_results") or []] 267 current_step = int(state.get("step_number", len(step_results) - 1)) + 1 268 269 ctx = BehaviorContext( 270 agent_id=self.agent_id, 271 task=task, 272 execution_id=execution_id, 273 last_completed_step=max(current_step - 1, -1), 274 metadata={"resume_agent": self}, 275 ) 276 ctx = await self._apply_behaviors_before(ctx) 277 278 for i in range(current_step, len(plan)): 279 step_num = i + 1 280 result_text = await self._execute_step(task, plan, step_num, step_results) 281 step_results.append(result_text) 282 agent_step = AgentStep( 283 thought=f"Executing plan step {step_num}: {plan[i]}", 284 action="llm_execution", 285 action_input={"step": plan[i]}, 286 observation=result_text, 287 ) 288 steps.append(agent_step) 289 await self._save_checkpoint( 290 execution_id, 291 task, 292 context, 293 plan, 294 steps, 295 step_results, 296 i, 297 ) 298 ctx.last_completed_step = i 299 300 final_output = step_results[-1] if step_results else "" 301 result = AgentResult( 302 output=final_output, 303 steps=steps, 304 success=True, 305 metadata={"plan": plan, "execution_id": execution_id}, 306 ) 307 return await self._apply_behaviors_after(ctx, result) 308 309 # ------------------------------------------------------------------ 310 # BaseAgent implementation 311 # ------------------------------------------------------------------ 312 313 async def execute( 314 self, task: str, context: Optional[Dict[str, Any]] = None 315 ) -> AgentResult: 316 execution_id = str(uuid4()) 317 run_context = context or {} 318 ctx = BehaviorContext( 319 agent_id=self.agent_id, 320 task=task, 321 execution_id=execution_id, 322 metadata={"resume_agent": self, **run_context}, 323 ) 324 ctx = await self._apply_behaviors_before(ctx) 325 self._log_execution_start(task) 326 327 with self._tracer.trace( 328 "plan_execute_agent.execute", input=task, metadata={"agent_id": self.agent_id} 329 ) as trace: 330 try: 331 # Phase 1 — Plan 332 with trace.span("plan", input=task) as plan_span: 333 plan = await self._plan(task) 334 plan_span.set_output({"steps": plan}) 335 self._logger.info( 336 "Plan created", 337 agent_id=self.agent_id, 338 step_count=len(plan), 339 ) 340 await self._save_checkpoint( 341 execution_id, 342 task, 343 run_context, 344 plan, 345 [], 346 [], 347 -1, 348 ) 349 350 # Phase 2 — Execute 351 steps: List[AgentStep] = [] 352 step_results: List[str] = [] 353 for i, plan_step in enumerate(plan): 354 step_num = i + 1 355 with trace.span(f"execute_step_{step_num}", input=plan_step) as step_span: 356 perf_id = None 357 if self._performance_monitor: 358 perf_id = self._performance_monitor.start_request( 359 provider="llm_gateway", 360 model=self.model or "default", 361 ) 362 try: 363 result_text = await self._execute_step( 364 task, plan, step_num, step_results 365 ) 366 if self._performance_monitor and perf_id: 367 self._performance_monitor.end_request( 368 request_id=perf_id, 369 prompt_tokens=0, 370 completion_tokens=0, 371 success=True, 372 ) 373 except Exception as exc: 374 if self._performance_monitor and perf_id: 375 self._performance_monitor.end_request( 376 request_id=perf_id, 377 prompt_tokens=0, 378 completion_tokens=0, 379 success=False, 380 error=str(exc), 381 ) 382 raise 383 384 step_results.append(result_text) 385 agent_step = AgentStep( 386 thought=f"Executing plan step {step_num}: {plan_step}", 387 action="llm_execution", 388 action_input={"step": plan_step}, 389 observation=result_text, 390 ) 391 steps.append(agent_step) 392 step_span.set_output(result_text) 393 await self._save_checkpoint( 394 execution_id, 395 task, 396 run_context, 397 plan, 398 steps, 399 step_results, 400 i, 401 ) 402 ctx.last_completed_step = i 403 404 final_output = step_results[-1] if step_results else "" 405 if self.state_store: 406 await self.state_store.set( 407 f"agent:{self.agent_id}:last_steps", 408 [vars(s) for s in steps], 409 ttl=3600, 410 ) 411 412 result = AgentResult( 413 output=final_output, 414 steps=steps, 415 success=True, 416 metadata={"plan": plan, "execution_id": execution_id}, 417 ) 418 result = await self._apply_behaviors_after(ctx, result) 419 self._log_execution_end(task, success=True, steps=len(steps)) 420 trace.set_output(final_output) 421 return result 422 423 except Exception as exc: 424 self._log_execution_error(task, exc) 425 fallback = await self._apply_behaviors_on_error(ctx, exc) 426 if fallback is not None: 427 return fallback 428 raise 429 430 async def stream_execute( 431 self, task: str, context: Optional[Dict[str, Any]] = None 432 ) -> AsyncIterator[AgentStep]: 433 ctx = BehaviorContext(agent_id=self.agent_id, task=task) 434 ctx = await self._apply_behaviors_before(ctx) 435 self._log_execution_start(task) 436 437 plan = await self._plan(task) 438 step_results: List[str] = [] 439 for i, plan_step in enumerate(plan): 440 result_text = await self._execute_step(task, plan, i + 1, step_results) 441 step_results.append(result_text) 442 step = AgentStep( 443 thought=f"Executing plan step {i+1}: {plan_step}", 444 action="llm_execution", 445 action_input={"step": plan_step}, 446 observation=result_text, 447 ) 448 yield step
Two-phase agent: LLM plans all steps first, then executes each step sequentially.
Phase 1 — Plan: Ask the LLM to decompose the task into a JSON list of steps.
Phase 2 — Execute: Feed each step back to the LLM (with accumulated context) to
produce a result. Each step runs an inner tool-calling loop — if the LLM emits a
Thought/Action/Action Input block referencing a registered tool, the tool is
invoked and its observation is fed back for the LLM to produce a final step result.
Args:
max_plan_steps: Maximum number of planned steps allowed (default: 10).
max_tool_calls_per_step: Maximum tool calls allowed within a single plan step
(default: 3).
model: LLM model name (optional).
temperature: Sampling temperature (default: 0.1).
plan_prompt: Override the planning prompt. Must contain {task}.
execute_prompt: Override the step-execution prompt. Must contain
{task}, {plan}, {previous_results}, {step_num},
{current_step}, and {tool_section}.
All other args inherited from BaseAgent.
PlanExecuteAgent( *args: Any, max_plan_steps: int = 10, max_tool_calls_per_step: int = 3, model: Optional[str] = None, temperature: float = 0.1, plan_prompt: Optional[str] = None, execute_prompt: Optional[str] = None, **kwargs: Any)
87 def __init__( 88 self, 89 *args: Any, 90 max_plan_steps: int = 10, 91 max_tool_calls_per_step: int = 3, 92 model: Optional[str] = None, 93 temperature: float = 0.1, 94 plan_prompt: Optional[str] = None, 95 execute_prompt: Optional[str] = None, 96 **kwargs: Any, 97 ) -> None: 98 super().__init__(*args, **kwargs) 99 self.max_plan_steps = max_plan_steps 100 self.max_tool_calls_per_step = max_tool_calls_per_step 101 self.model = model 102 self.temperature = temperature 103 self._plan_prompt = plan_prompt 104 self._execute_prompt = execute_prompt
DEFAULT_PLAN_PROMPT: str =
'You are a planning assistant. Given a task, decompose it into a numbered\nlist of clear, sequential steps. Each step should be a single actionable instruction.\n\nTask: {task}\n\nRespond with ONLY a JSON array of step strings:\n["step 1 description", "step 2 description", ...]\n'
DEFAULT_EXECUTE_PROMPT: str =
'You are executing step {step_num} of a multi-step plan.\n\nOriginal task: {task}\nFull plan: {plan}\nPrevious results:\n{previous_results}\n\nCurrent step: {current_step}\n\n{tool_section}Complete this step and provide the result. Be concise and factual.\n\nIf you need to use a tool, respond with:\nThought: <your reasoning>\nAction: <tool name>\nAction Input: <JSON object with tool arguments>\n\nOtherwise, just provide the result directly.\n'
240 async def resume_from( 241 self, execution_id: str, from_step: Optional[int] = None 242 ) -> AgentResult: 243 """Resume a Plan-Execute run from latest (or selected) checkpoint.""" 244 if not self.checkpoint_manager: 245 raise ValueError("Checkpoint manager is not configured for this agent") 246 247 checkpoint = None 248 if from_step is not None: 249 checkpoints = await self.checkpoint_manager.list_by_execution(execution_id) 250 for ckpt in checkpoints: 251 if ckpt.state.get("step_number") == from_step: 252 checkpoint = ckpt 253 break 254 else: 255 checkpoint = await self.checkpoint_manager.load_latest_for_execution(execution_id) 256 257 if checkpoint is None: 258 raise ValueError(f"No checkpoint found for execution_id={execution_id}") 259 260 state = checkpoint.state 261 task = str(state.get("task", "")) 262 context = state.get("context") or {} 263 plan = [str(s) for s in state.get("plan") or []] 264 raw_steps = state.get("steps") or [] 265 steps = [AgentStep(**raw) for raw in raw_steps if isinstance(raw, dict)] 266 step_results = [str(r) for r in state.get("step_results") or []] 267 current_step = int(state.get("step_number", len(step_results) - 1)) + 1 268 269 ctx = BehaviorContext( 270 agent_id=self.agent_id, 271 task=task, 272 execution_id=execution_id, 273 last_completed_step=max(current_step - 1, -1), 274 metadata={"resume_agent": self}, 275 ) 276 ctx = await self._apply_behaviors_before(ctx) 277 278 for i in range(current_step, len(plan)): 279 step_num = i + 1 280 result_text = await self._execute_step(task, plan, step_num, step_results) 281 step_results.append(result_text) 282 agent_step = AgentStep( 283 thought=f"Executing plan step {step_num}: {plan[i]}", 284 action="llm_execution", 285 action_input={"step": plan[i]}, 286 observation=result_text, 287 ) 288 steps.append(agent_step) 289 await self._save_checkpoint( 290 execution_id, 291 task, 292 context, 293 plan, 294 steps, 295 step_results, 296 i, 297 ) 298 ctx.last_completed_step = i 299 300 final_output = step_results[-1] if step_results else "" 301 result = AgentResult( 302 output=final_output, 303 steps=steps, 304 success=True, 305 metadata={"plan": plan, "execution_id": execution_id}, 306 ) 307 return await self._apply_behaviors_after(ctx, result)
Resume a Plan-Execute run from latest (or selected) checkpoint.
313 async def execute( 314 self, task: str, context: Optional[Dict[str, Any]] = None 315 ) -> AgentResult: 316 execution_id = str(uuid4()) 317 run_context = context or {} 318 ctx = BehaviorContext( 319 agent_id=self.agent_id, 320 task=task, 321 execution_id=execution_id, 322 metadata={"resume_agent": self, **run_context}, 323 ) 324 ctx = await self._apply_behaviors_before(ctx) 325 self._log_execution_start(task) 326 327 with self._tracer.trace( 328 "plan_execute_agent.execute", input=task, metadata={"agent_id": self.agent_id} 329 ) as trace: 330 try: 331 # Phase 1 — Plan 332 with trace.span("plan", input=task) as plan_span: 333 plan = await self._plan(task) 334 plan_span.set_output({"steps": plan}) 335 self._logger.info( 336 "Plan created", 337 agent_id=self.agent_id, 338 step_count=len(plan), 339 ) 340 await self._save_checkpoint( 341 execution_id, 342 task, 343 run_context, 344 plan, 345 [], 346 [], 347 -1, 348 ) 349 350 # Phase 2 — Execute 351 steps: List[AgentStep] = [] 352 step_results: List[str] = [] 353 for i, plan_step in enumerate(plan): 354 step_num = i + 1 355 with trace.span(f"execute_step_{step_num}", input=plan_step) as step_span: 356 perf_id = None 357 if self._performance_monitor: 358 perf_id = self._performance_monitor.start_request( 359 provider="llm_gateway", 360 model=self.model or "default", 361 ) 362 try: 363 result_text = await self._execute_step( 364 task, plan, step_num, step_results 365 ) 366 if self._performance_monitor and perf_id: 367 self._performance_monitor.end_request( 368 request_id=perf_id, 369 prompt_tokens=0, 370 completion_tokens=0, 371 success=True, 372 ) 373 except Exception as exc: 374 if self._performance_monitor and perf_id: 375 self._performance_monitor.end_request( 376 request_id=perf_id, 377 prompt_tokens=0, 378 completion_tokens=0, 379 success=False, 380 error=str(exc), 381 ) 382 raise 383 384 step_results.append(result_text) 385 agent_step = AgentStep( 386 thought=f"Executing plan step {step_num}: {plan_step}", 387 action="llm_execution", 388 action_input={"step": plan_step}, 389 observation=result_text, 390 ) 391 steps.append(agent_step) 392 step_span.set_output(result_text) 393 await self._save_checkpoint( 394 execution_id, 395 task, 396 run_context, 397 plan, 398 steps, 399 step_results, 400 i, 401 ) 402 ctx.last_completed_step = i 403 404 final_output = step_results[-1] if step_results else "" 405 if self.state_store: 406 await self.state_store.set( 407 f"agent:{self.agent_id}:last_steps", 408 [vars(s) for s in steps], 409 ttl=3600, 410 ) 411 412 result = AgentResult( 413 output=final_output, 414 steps=steps, 415 success=True, 416 metadata={"plan": plan, "execution_id": execution_id}, 417 ) 418 result = await self._apply_behaviors_after(ctx, result) 419 self._log_execution_end(task, success=True, steps=len(steps)) 420 trace.set_output(final_output) 421 return result 422 423 except Exception as exc: 424 self._log_execution_error(task, exc) 425 fallback = await self._apply_behaviors_on_error(ctx, exc) 426 if fallback is not None: 427 return fallback 428 raise
Execute the task and return a result.
async def
stream_execute( self, task: str, context: Optional[Dict[str, Any]] = None) -> AsyncIterator[AgentStep]:
430 async def stream_execute( 431 self, task: str, context: Optional[Dict[str, Any]] = None 432 ) -> AsyncIterator[AgentStep]: 433 ctx = BehaviorContext(agent_id=self.agent_id, task=task) 434 ctx = await self._apply_behaviors_before(ctx) 435 self._log_execution_start(task) 436 437 plan = await self._plan(task) 438 step_results: List[str] = [] 439 for i, plan_step in enumerate(plan): 440 result_text = await self._execute_step(task, plan, i + 1, step_results) 441 step_results.append(result_text) 442 step = AgentStep( 443 thought=f"Executing plan step {i+1}: {plan_step}", 444 action="llm_execution", 445 action_input={"step": plan_step}, 446 observation=result_text, 447 ) 448 yield step
Execute the task, yielding each step as it completes.
31class ReflexionAgent(BaseAgent): 32 """ 33 Wraps another agent and applies self-reflection on failure. 34 35 After each attempt, a separate LLM call critiques the output. If the 36 critique says the result is unsatisfactory, the agent reflects and retries 37 with an improved prompt up to ``max_reflections`` times. 38 39 Args: 40 inner_agent: The underlying agent to run and reflect on. 41 max_reflections: Maximum reflection/retry cycles (default: 2). 42 model: Model for critique and reflection calls (defaults to inner agent 43 model or gateway default). 44 critique_prompt: Override the critique prompt. Must contain ``{task}`` 45 and ``{response}``. 46 reflect_prompt: Override the reflection/retry prompt. Must contain 47 ``{task}``, ``{previous_output}``, and ``{critique}``. 48 49 All other args inherited from :class:`BaseAgent`. 50 """ 51 52 def __init__( 53 self, 54 inner_agent: BaseAgent, 55 *args: Any, 56 max_reflections: int = 2, 57 model: Optional[str] = None, 58 critique_prompt: Optional[str] = None, 59 reflect_prompt: Optional[str] = None, 60 **kwargs: Any, 61 ) -> None: 62 super().__init__(inner_agent.llm_gateway, *args, **kwargs) 63 self.inner_agent = inner_agent 64 self.max_reflections = max_reflections 65 self.model = model 66 self._critique_prompt = critique_prompt 67 self._reflect_prompt = reflect_prompt 68 69 #: Default critique prompt. Must contain ``{task}`` and ``{response}`` if overriding. 70 DEFAULT_CRITIQUE_PROMPT: str = _CRITIQUE_PROMPT 71 #: Default reflection prompt. Must contain ``{task}``, ``{previous_output}``, 72 #: and ``{critique}`` if overriding. 73 DEFAULT_REFLECT_PROMPT: str = _REFLECT_PROMPT 74 75 async def _critique(self, task: str, output: str) -> Optional[str]: 76 """Returns the critique reason string, or None if the output is satisfactory.""" 77 template = self._critique_prompt if self._critique_prompt is not None else _CRITIQUE_PROMPT 78 prompt = template.format(task=task, response=output) 79 response = await self.llm_gateway.complete(prompt, model=self.model, temperature=0.0) 80 text = response.content.strip() 81 if text.upper().startswith("YES"): 82 return None 83 # Extract reason after "NO:" 84 if ":" in text: 85 return text.split(":", 1)[1].strip() 86 return text 87 88 async def _reflect_and_retry( 89 self, task: str, previous_output: str, critique: str 90 ) -> AgentResult: 91 template = self._reflect_prompt if self._reflect_prompt is not None else _REFLECT_PROMPT 92 improved_task = template.format( 93 task=task, previous_output=previous_output, critique=critique 94 ) 95 return await self.inner_agent.execute(improved_task) 96 97 async def _save_checkpoint( 98 self, 99 execution_id: str, 100 task: str, 101 context: Dict[str, Any], 102 reflection_round: int, 103 result: AgentResult, 104 critique: Optional[str] = None, 105 ) -> None: 106 if not self.checkpoint_manager: 107 return 108 await self.checkpoint_manager.save( 109 agent_id=self.agent_id, 110 execution_id=execution_id, 111 state={ 112 "execution_id": execution_id, 113 "task": task, 114 "context": context, 115 "reflection_round": reflection_round, 116 "critique": critique, 117 "result": { 118 "output": result.output, 119 "success": result.success, 120 "error": result.error, 121 "metadata": result.metadata, 122 "steps": [dataclasses.asdict(s) for s in result.steps], 123 }, 124 }, 125 metadata={"reflection_round": reflection_round}, 126 ttl=86400, 127 ) 128 129 async def resume_from( 130 self, execution_id: str, from_reflection: Optional[int] = None 131 ) -> AgentResult: 132 """Resume a reflexion execution from latest (or selected) reflection checkpoint.""" 133 if not self.checkpoint_manager: 134 raise ValueError("Checkpoint manager is not configured for this agent") 135 136 checkpoint = None 137 if from_reflection is not None: 138 checkpoints = await self.checkpoint_manager.list_by_execution(execution_id) 139 for ckpt in checkpoints: 140 if ckpt.state.get("reflection_round") == from_reflection: 141 checkpoint = ckpt 142 break 143 else: 144 checkpoint = await self.checkpoint_manager.load_latest_for_execution(execution_id) 145 146 if checkpoint is None: 147 raise ValueError(f"No checkpoint found for execution_id={execution_id}") 148 149 state = checkpoint.state 150 task = str(state.get("task", "")) 151 context = state.get("context") or {} 152 reflection_round = int(state.get("reflection_round", 0)) 153 raw = state.get("result") or {} 154 155 steps = [ 156 AgentStep(**s) for s in raw.get("steps", []) if isinstance(s, dict) 157 ] 158 result = AgentResult( 159 output=str(raw.get("output", "")), 160 steps=steps, 161 success=bool(raw.get("success", True)), 162 error=raw.get("error"), 163 metadata=raw.get("metadata", {}), 164 ) 165 166 ctx = BehaviorContext( 167 agent_id=self.agent_id, 168 task=task, 169 execution_id=execution_id, 170 last_completed_step=reflection_round, 171 metadata={"resume_agent": self}, 172 ) 173 ctx = await self._apply_behaviors_before(ctx) 174 175 all_steps: List[AgentStep] = list(result.steps) 176 for idx in range(reflection_round, self.max_reflections): 177 critique = await self._critique(task, result.output) 178 if critique is None: 179 break 180 result = await self._reflect_and_retry(task, result.output, critique) 181 all_steps.extend(result.steps) 182 await self._save_checkpoint( 183 execution_id, 184 task, 185 context, 186 idx + 1, 187 result, 188 critique, 189 ) 190 ctx.last_completed_step = idx + 1 191 192 final = AgentResult( 193 output=result.output, 194 steps=all_steps, 195 success=result.success, 196 error=result.error, 197 metadata={**result.metadata, "execution_id": execution_id}, 198 ) 199 return await self._apply_behaviors_after(ctx, final) 200 201 # ------------------------------------------------------------------ 202 # BaseAgent implementation 203 # ------------------------------------------------------------------ 204 205 async def execute( 206 self, task: str, context: Optional[Dict[str, Any]] = None 207 ) -> AgentResult: 208 execution_id = str(uuid4()) 209 run_context = context or {} 210 ctx = BehaviorContext( 211 agent_id=self.agent_id, 212 task=task, 213 execution_id=execution_id, 214 metadata={"resume_agent": self, **run_context}, 215 ) 216 ctx = await self._apply_behaviors_before(ctx) 217 self._log_execution_start(task) 218 219 with self._tracer.trace( 220 "reflexion_agent.execute", input=task, metadata={"agent_id": self.agent_id} 221 ) as trace: 222 try: 223 result = await self.inner_agent.execute(task, context) 224 all_steps: List[AgentStep] = list(result.steps) 225 await self._save_checkpoint(execution_id, task, run_context, 0, result) 226 ctx.last_completed_step = 0 227 228 for reflection_num in range(self.max_reflections): 229 with trace.span( 230 f"critique_{reflection_num + 1}", input=result.output 231 ) as cspan: 232 critique = await self._critique(task, result.output) 233 cspan.set_output({"satisfactory": critique is None, "critique": critique}) 234 235 if critique is None: 236 self._logger.info( 237 "Reflexion: output accepted", 238 agent_id=self.agent_id, 239 reflection=reflection_num + 1, 240 ) 241 break 242 243 self._logger.info( 244 "Reflexion: retrying", 245 agent_id=self.agent_id, 246 reflection=reflection_num + 1, 247 critique=critique, 248 ) 249 if self._metrics: 250 self._metrics.increment( 251 "agent.reflexions", agent_id=self.agent_id 252 ) 253 254 with trace.span(f"reflect_{reflection_num + 1}", input=critique) as rspan: 255 result = await self._reflect_and_retry(task, result.output, critique) 256 all_steps.extend(result.steps) 257 await self._save_checkpoint( 258 execution_id, 259 task, 260 run_context, 261 reflection_num + 1, 262 result, 263 critique, 264 ) 265 ctx.last_completed_step = reflection_num + 1 266 rspan.set_output(result.output) 267 268 final = AgentResult( 269 output=result.output, 270 steps=all_steps, 271 success=result.success, 272 error=result.error, 273 metadata={**result.metadata, "execution_id": execution_id}, 274 ) 275 final = await self._apply_behaviors_after(ctx, final) 276 self._log_execution_end(task, success=final.success, steps=len(all_steps)) 277 trace.set_output(final.output) 278 return final 279 280 except Exception as exc: 281 self._log_execution_error(task, exc) 282 fallback = await self._apply_behaviors_on_error(ctx, exc) 283 if fallback is not None: 284 return fallback 285 raise 286 287 async def stream_execute( 288 self, task: str, context: Optional[Dict[str, Any]] = None 289 ) -> AsyncIterator[AgentStep]: 290 # Reflexion does full-cycle reflection, so streaming yields inner steps 291 async for step in self.inner_agent.stream_execute(task, context): 292 yield step
Wraps another agent and applies self-reflection on failure.
After each attempt, a separate LLM call critiques the output. If the
critique says the result is unsatisfactory, the agent reflects and retries
with an improved prompt up to max_reflections times.
Args:
inner_agent: The underlying agent to run and reflect on.
max_reflections: Maximum reflection/retry cycles (default: 2).
model: Model for critique and reflection calls (defaults to inner agent
model or gateway default).
critique_prompt: Override the critique prompt. Must contain {task}
and {response}.
reflect_prompt: Override the reflection/retry prompt. Must contain
{task}, {previous_output}, and {critique}.
All other args inherited from BaseAgent.
ReflexionAgent( inner_agent: BaseAgent, *args: Any, max_reflections: int = 2, model: Optional[str] = None, critique_prompt: Optional[str] = None, reflect_prompt: Optional[str] = None, **kwargs: Any)
52 def __init__( 53 self, 54 inner_agent: BaseAgent, 55 *args: Any, 56 max_reflections: int = 2, 57 model: Optional[str] = None, 58 critique_prompt: Optional[str] = None, 59 reflect_prompt: Optional[str] = None, 60 **kwargs: Any, 61 ) -> None: 62 super().__init__(inner_agent.llm_gateway, *args, **kwargs) 63 self.inner_agent = inner_agent 64 self.max_reflections = max_reflections 65 self.model = model 66 self._critique_prompt = critique_prompt 67 self._reflect_prompt = reflect_prompt
DEFAULT_CRITIQUE_PROMPT: str =
'Evaluate the following response to the task.\n\nTask: {task}\nResponse: {response}\n\nIs this response satisfactory? Reply with ONLY:\n- "YES" if the response fully and correctly addresses the task.\n- "NO: <brief reason>" if it does not.\n'
DEFAULT_REFLECT_PROMPT: str =
'You just attempted a task and the result was unsatisfactory.\n\nTask: {task}\nPrevious attempt output: {previous_output}\nReason it was unsatisfactory: {critique}\n\nReflect on what went wrong and provide an improved, more thorough answer.\n'
async def
resume_from( self, execution_id: str, from_reflection: Optional[int] = None) -> AgentResult:
129 async def resume_from( 130 self, execution_id: str, from_reflection: Optional[int] = None 131 ) -> AgentResult: 132 """Resume a reflexion execution from latest (or selected) reflection checkpoint.""" 133 if not self.checkpoint_manager: 134 raise ValueError("Checkpoint manager is not configured for this agent") 135 136 checkpoint = None 137 if from_reflection is not None: 138 checkpoints = await self.checkpoint_manager.list_by_execution(execution_id) 139 for ckpt in checkpoints: 140 if ckpt.state.get("reflection_round") == from_reflection: 141 checkpoint = ckpt 142 break 143 else: 144 checkpoint = await self.checkpoint_manager.load_latest_for_execution(execution_id) 145 146 if checkpoint is None: 147 raise ValueError(f"No checkpoint found for execution_id={execution_id}") 148 149 state = checkpoint.state 150 task = str(state.get("task", "")) 151 context = state.get("context") or {} 152 reflection_round = int(state.get("reflection_round", 0)) 153 raw = state.get("result") or {} 154 155 steps = [ 156 AgentStep(**s) for s in raw.get("steps", []) if isinstance(s, dict) 157 ] 158 result = AgentResult( 159 output=str(raw.get("output", "")), 160 steps=steps, 161 success=bool(raw.get("success", True)), 162 error=raw.get("error"), 163 metadata=raw.get("metadata", {}), 164 ) 165 166 ctx = BehaviorContext( 167 agent_id=self.agent_id, 168 task=task, 169 execution_id=execution_id, 170 last_completed_step=reflection_round, 171 metadata={"resume_agent": self}, 172 ) 173 ctx = await self._apply_behaviors_before(ctx) 174 175 all_steps: List[AgentStep] = list(result.steps) 176 for idx in range(reflection_round, self.max_reflections): 177 critique = await self._critique(task, result.output) 178 if critique is None: 179 break 180 result = await self._reflect_and_retry(task, result.output, critique) 181 all_steps.extend(result.steps) 182 await self._save_checkpoint( 183 execution_id, 184 task, 185 context, 186 idx + 1, 187 result, 188 critique, 189 ) 190 ctx.last_completed_step = idx + 1 191 192 final = AgentResult( 193 output=result.output, 194 steps=all_steps, 195 success=result.success, 196 error=result.error, 197 metadata={**result.metadata, "execution_id": execution_id}, 198 ) 199 return await self._apply_behaviors_after(ctx, final)
Resume a reflexion execution from latest (or selected) reflection checkpoint.
205 async def execute( 206 self, task: str, context: Optional[Dict[str, Any]] = None 207 ) -> AgentResult: 208 execution_id = str(uuid4()) 209 run_context = context or {} 210 ctx = BehaviorContext( 211 agent_id=self.agent_id, 212 task=task, 213 execution_id=execution_id, 214 metadata={"resume_agent": self, **run_context}, 215 ) 216 ctx = await self._apply_behaviors_before(ctx) 217 self._log_execution_start(task) 218 219 with self._tracer.trace( 220 "reflexion_agent.execute", input=task, metadata={"agent_id": self.agent_id} 221 ) as trace: 222 try: 223 result = await self.inner_agent.execute(task, context) 224 all_steps: List[AgentStep] = list(result.steps) 225 await self._save_checkpoint(execution_id, task, run_context, 0, result) 226 ctx.last_completed_step = 0 227 228 for reflection_num in range(self.max_reflections): 229 with trace.span( 230 f"critique_{reflection_num + 1}", input=result.output 231 ) as cspan: 232 critique = await self._critique(task, result.output) 233 cspan.set_output({"satisfactory": critique is None, "critique": critique}) 234 235 if critique is None: 236 self._logger.info( 237 "Reflexion: output accepted", 238 agent_id=self.agent_id, 239 reflection=reflection_num + 1, 240 ) 241 break 242 243 self._logger.info( 244 "Reflexion: retrying", 245 agent_id=self.agent_id, 246 reflection=reflection_num + 1, 247 critique=critique, 248 ) 249 if self._metrics: 250 self._metrics.increment( 251 "agent.reflexions", agent_id=self.agent_id 252 ) 253 254 with trace.span(f"reflect_{reflection_num + 1}", input=critique) as rspan: 255 result = await self._reflect_and_retry(task, result.output, critique) 256 all_steps.extend(result.steps) 257 await self._save_checkpoint( 258 execution_id, 259 task, 260 run_context, 261 reflection_num + 1, 262 result, 263 critique, 264 ) 265 ctx.last_completed_step = reflection_num + 1 266 rspan.set_output(result.output) 267 268 final = AgentResult( 269 output=result.output, 270 steps=all_steps, 271 success=result.success, 272 error=result.error, 273 metadata={**result.metadata, "execution_id": execution_id}, 274 ) 275 final = await self._apply_behaviors_after(ctx, final) 276 self._log_execution_end(task, success=final.success, steps=len(all_steps)) 277 trace.set_output(final.output) 278 return final 279 280 except Exception as exc: 281 self._log_execution_error(task, exc) 282 fallback = await self._apply_behaviors_on_error(ctx, exc) 283 if fallback is not None: 284 return fallback 285 raise
Execute the task and return a result.
async def
stream_execute( self, task: str, context: Optional[Dict[str, Any]] = None) -> AsyncIterator[AgentStep]:
287 async def stream_execute( 288 self, task: str, context: Optional[Dict[str, Any]] = None 289 ) -> AsyncIterator[AgentStep]: 290 # Reflexion does full-cycle reflection, so streaming yields inner steps 291 async for step in self.inner_agent.stream_execute(task, context): 292 yield step
Execute the task, yielding each step as it completes.
28class ChainOfThoughtAgent(BaseAgent): 29 """ 30 Prompts the LLM to reason via a structured ``<thinking>`` scratchpad before 31 producing a final answer. 32 33 **When to use:** Tasks requiring structured multi-step reasoning with no 34 external tool calls — classification, analysis, summarisation, math, or 35 any question answerable from the LLM's own knowledge. Single LLM call: 36 lower latency and cost than :class:`ReActAgent`. 37 38 **When NOT to use:** Tasks that require searching a database, calling an 39 API, or any live data lookup — use :class:`ReActAgent` instead. 40 41 .. note:: 42 Passing a ``tool_registry`` to this agent has no effect — tools are 43 never invoked. If you need tool calls, use :class:`ReActAgent`. 44 45 The scratchpad is extracted and stored in the returned :class:`AgentStep` 46 as the ``thought`` field, while the final answer becomes ``observation``. 47 48 Args: 49 model: LLM model name (optional). 50 temperature: Sampling temperature (default: 0.1). 51 system_prompt: Override the default Chain-of-Thought prompt template. 52 Must contain ``{task}`` as a placeholder where the task will be 53 inserted. 54 55 All other args inherited from :class:`BaseAgent`. 56 """ 57 58 def __init__( 59 self, 60 *args: Any, 61 model: Optional[str] = None, 62 temperature: float = 0.1, 63 system_prompt: Optional[str] = None, 64 **kwargs: Any, 65 ) -> None: 66 super().__init__(*args, **kwargs) 67 self.model = model 68 self.temperature = temperature 69 self._system_prompt = system_prompt 70 71 #: Default CoT prompt template used when no ``system_prompt`` is passed. 72 #: Must contain ``{task}`` if overriding. 73 DEFAULT_SYSTEM_PROMPT: str = _COT_PROMPT 74 75 def _build_prompt(self, task: str) -> str: 76 template = self._system_prompt if self._system_prompt is not None else _COT_PROMPT 77 return template.format(task=task) 78 79 def _parse_response(self, text: str) -> tuple[str, str]: 80 """Returns (thinking_scratchpad, final_answer).""" 81 match = _THINKING_PATTERN.search(text) 82 if match: 83 thinking = match.group(1).strip() 84 answer = _THINKING_PATTERN.sub("", text).strip() 85 else: 86 # No <thinking> block — treat everything as the answer 87 thinking = "" 88 answer = text.strip() 89 return thinking, answer 90 91 async def _save_checkpoint( 92 self, 93 execution_id: str, 94 task: str, 95 context: Dict[str, Any], 96 thinking: str, 97 answer: str, 98 ) -> None: 99 if not self.checkpoint_manager: 100 return 101 await self.checkpoint_manager.save( 102 agent_id=self.agent_id, 103 execution_id=execution_id, 104 state={ 105 "execution_id": execution_id, 106 "task": task, 107 "context": context, 108 "step_number": 0, 109 "steps": [ 110 { 111 "thought": thinking, 112 "action": "chain_of_thought", 113 "action_input": {"task": task}, 114 "observation": answer, 115 "metadata": {}, 116 } 117 ], 118 "thinking": thinking, 119 "answer": answer, 120 }, 121 metadata={"step_number": 0}, 122 ttl=86400, 123 ) 124 125 async def resume_from(self, execution_id: str) -> AgentResult: 126 """Resume (idempotently) from a completed Chain-of-Thought checkpoint.""" 127 if not self.checkpoint_manager: 128 raise ValueError("Checkpoint manager is not configured for this agent") 129 130 checkpoint = await self.checkpoint_manager.load_latest_for_execution(execution_id) 131 if checkpoint is None: 132 raise ValueError(f"No checkpoint found for execution_id={execution_id}") 133 134 state = checkpoint.state 135 thinking = str(state.get("thinking", "")) 136 answer = str(state.get("answer", "")) 137 task = str(state.get("task", "")) 138 139 ctx = BehaviorContext( 140 agent_id=self.agent_id, 141 task=task, 142 execution_id=execution_id, 143 last_completed_step=0, 144 metadata={"resume_agent": self}, 145 ) 146 ctx = await self._apply_behaviors_before(ctx) 147 148 step = AgentStep( 149 thought=thinking, 150 action="chain_of_thought", 151 action_input={"task": task}, 152 observation=answer, 153 ) 154 result = AgentResult( 155 output=answer, 156 steps=[step], 157 success=True, 158 metadata={"thinking": thinking, "execution_id": execution_id}, 159 ) 160 return await self._apply_behaviors_after(ctx, result) 161 162 # ------------------------------------------------------------------ 163 # BaseAgent implementation 164 # ------------------------------------------------------------------ 165 166 async def execute( 167 self, task: str, context: Optional[Dict[str, Any]] = None 168 ) -> AgentResult: 169 execution_id = str(uuid4()) 170 run_context = context or {} 171 ctx = BehaviorContext( 172 agent_id=self.agent_id, 173 task=task, 174 execution_id=execution_id, 175 metadata={"resume_agent": self, **run_context}, 176 ) 177 ctx = await self._apply_behaviors_before(ctx) 178 self._log_execution_start(task) 179 180 with self._tracer.trace( 181 "cot_agent.execute", input=task, metadata={"agent_id": self.agent_id} 182 ) as trace: 183 try: 184 prompt = self._build_prompt(task) 185 186 perf_id = None 187 if self._performance_monitor: 188 perf_id = self._performance_monitor.start_request( 189 provider="llm_gateway", model=self.model or "default" 190 ) 191 192 with trace.generation("llm_call", model=self.model or "default", input=prompt) as gen: 193 try: 194 response = await self.llm_gateway.complete( 195 prompt, model=self.model, temperature=self.temperature 196 ) 197 if self._performance_monitor and perf_id: 198 self._performance_monitor.end_request( 199 request_id=perf_id, 200 prompt_tokens=response.usage.get("prompt_tokens", 0), 201 completion_tokens=response.usage.get("completion_tokens", 0), 202 success=True, 203 ) 204 gen.set_output(response.content) 205 gen.set_token_usage(**response.usage) 206 except Exception as exc: 207 if self._performance_monitor and perf_id: 208 self._performance_monitor.end_request( 209 request_id=perf_id, 210 prompt_tokens=0, 211 completion_tokens=0, 212 success=False, 213 error=str(exc), 214 ) 215 raise 216 217 thinking, answer = self._parse_response(response.content) 218 step = AgentStep( 219 thought=thinking, 220 action="chain_of_thought", 221 action_input={"task": task}, 222 observation=answer, 223 ) 224 225 if self.state_store: 226 await self.state_store.set( 227 f"agent:{self.agent_id}:last_steps", 228 [vars(step)], 229 ttl=3600, 230 ) 231 232 await self._save_checkpoint( 233 execution_id=execution_id, 234 task=task, 235 context=run_context, 236 thinking=thinking, 237 answer=answer, 238 ) 239 ctx.last_completed_step = 0 240 241 result = AgentResult( 242 output=answer, 243 steps=[step], 244 success=True, 245 metadata={"thinking": thinking, "execution_id": execution_id}, 246 ) 247 result = await self._apply_behaviors_after(ctx, result) 248 self._log_execution_end(task, success=True, steps=1) 249 trace.set_output(answer) 250 return result 251 252 except Exception as exc: 253 self._log_execution_error(task, exc) 254 fallback = await self._apply_behaviors_on_error(ctx, exc) 255 if fallback is not None: 256 return fallback 257 raise 258 259 async def stream_execute( 260 self, task: str, context: Optional[Dict[str, Any]] = None 261 ) -> AsyncIterator[AgentStep]: 262 result = await self.execute(task, context) 263 for step in result.steps: 264 yield step
Prompts the LLM to reason via a structured <thinking> scratchpad before
producing a final answer.
When to use: Tasks requiring structured multi-step reasoning with no
external tool calls — classification, analysis, summarisation, math, or
any question answerable from the LLM's own knowledge. Single LLM call:
lower latency and cost than ReActAgent.
When NOT to use: Tasks that require searching a database, calling an
API, or any live data lookup — use ReActAgent instead.
Passing a tool_registry to this agent has no effect — tools are
never invoked. If you need tool calls, use ReActAgent.
The scratchpad is extracted and stored in the returned AgentStep
as the thought field, while the final answer becomes observation.
Args:
model: LLM model name (optional).
temperature: Sampling temperature (default: 0.1).
system_prompt: Override the default Chain-of-Thought prompt template.
Must contain {task} as a placeholder where the task will be
inserted.
All other args inherited from BaseAgent.
ChainOfThoughtAgent( *args: Any, model: Optional[str] = None, temperature: float = 0.1, system_prompt: Optional[str] = None, **kwargs: Any)
58 def __init__( 59 self, 60 *args: Any, 61 model: Optional[str] = None, 62 temperature: float = 0.1, 63 system_prompt: Optional[str] = None, 64 **kwargs: Any, 65 ) -> None: 66 super().__init__(*args, **kwargs) 67 self.model = model 68 self.temperature = temperature 69 self._system_prompt = system_prompt
DEFAULT_SYSTEM_PROMPT: str =
'You are a careful, step-by-step reasoning assistant.\n\nBefore answering, work through the problem thoroughly inside <thinking> tags.\nThen provide your final answer outside the tags.\n\nFormat:\n<thinking>\n[Your detailed step-by-step reasoning here]\n</thinking>\n[Your final, concise answer here]\n\nTask: {task}\n'
125 async def resume_from(self, execution_id: str) -> AgentResult: 126 """Resume (idempotently) from a completed Chain-of-Thought checkpoint.""" 127 if not self.checkpoint_manager: 128 raise ValueError("Checkpoint manager is not configured for this agent") 129 130 checkpoint = await self.checkpoint_manager.load_latest_for_execution(execution_id) 131 if checkpoint is None: 132 raise ValueError(f"No checkpoint found for execution_id={execution_id}") 133 134 state = checkpoint.state 135 thinking = str(state.get("thinking", "")) 136 answer = str(state.get("answer", "")) 137 task = str(state.get("task", "")) 138 139 ctx = BehaviorContext( 140 agent_id=self.agent_id, 141 task=task, 142 execution_id=execution_id, 143 last_completed_step=0, 144 metadata={"resume_agent": self}, 145 ) 146 ctx = await self._apply_behaviors_before(ctx) 147 148 step = AgentStep( 149 thought=thinking, 150 action="chain_of_thought", 151 action_input={"task": task}, 152 observation=answer, 153 ) 154 result = AgentResult( 155 output=answer, 156 steps=[step], 157 success=True, 158 metadata={"thinking": thinking, "execution_id": execution_id}, 159 ) 160 return await self._apply_behaviors_after(ctx, result)
Resume (idempotently) from a completed Chain-of-Thought checkpoint.
166 async def execute( 167 self, task: str, context: Optional[Dict[str, Any]] = None 168 ) -> AgentResult: 169 execution_id = str(uuid4()) 170 run_context = context or {} 171 ctx = BehaviorContext( 172 agent_id=self.agent_id, 173 task=task, 174 execution_id=execution_id, 175 metadata={"resume_agent": self, **run_context}, 176 ) 177 ctx = await self._apply_behaviors_before(ctx) 178 self._log_execution_start(task) 179 180 with self._tracer.trace( 181 "cot_agent.execute", input=task, metadata={"agent_id": self.agent_id} 182 ) as trace: 183 try: 184 prompt = self._build_prompt(task) 185 186 perf_id = None 187 if self._performance_monitor: 188 perf_id = self._performance_monitor.start_request( 189 provider="llm_gateway", model=self.model or "default" 190 ) 191 192 with trace.generation("llm_call", model=self.model or "default", input=prompt) as gen: 193 try: 194 response = await self.llm_gateway.complete( 195 prompt, model=self.model, temperature=self.temperature 196 ) 197 if self._performance_monitor and perf_id: 198 self._performance_monitor.end_request( 199 request_id=perf_id, 200 prompt_tokens=response.usage.get("prompt_tokens", 0), 201 completion_tokens=response.usage.get("completion_tokens", 0), 202 success=True, 203 ) 204 gen.set_output(response.content) 205 gen.set_token_usage(**response.usage) 206 except Exception as exc: 207 if self._performance_monitor and perf_id: 208 self._performance_monitor.end_request( 209 request_id=perf_id, 210 prompt_tokens=0, 211 completion_tokens=0, 212 success=False, 213 error=str(exc), 214 ) 215 raise 216 217 thinking, answer = self._parse_response(response.content) 218 step = AgentStep( 219 thought=thinking, 220 action="chain_of_thought", 221 action_input={"task": task}, 222 observation=answer, 223 ) 224 225 if self.state_store: 226 await self.state_store.set( 227 f"agent:{self.agent_id}:last_steps", 228 [vars(step)], 229 ttl=3600, 230 ) 231 232 await self._save_checkpoint( 233 execution_id=execution_id, 234 task=task, 235 context=run_context, 236 thinking=thinking, 237 answer=answer, 238 ) 239 ctx.last_completed_step = 0 240 241 result = AgentResult( 242 output=answer, 243 steps=[step], 244 success=True, 245 metadata={"thinking": thinking, "execution_id": execution_id}, 246 ) 247 result = await self._apply_behaviors_after(ctx, result) 248 self._log_execution_end(task, success=True, steps=1) 249 trace.set_output(answer) 250 return result 251 252 except Exception as exc: 253 self._log_execution_error(task, exc) 254 fallback = await self._apply_behaviors_on_error(ctx, exc) 255 if fallback is not None: 256 return fallback 257 raise
Execute the task and return a result.
async def
stream_execute( self, task: str, context: Optional[Dict[str, Any]] = None) -> AsyncIterator[AgentStep]:
259 async def stream_execute( 260 self, task: str, context: Optional[Dict[str, Any]] = None 261 ) -> AsyncIterator[AgentStep]: 262 result = await self.execute(task, context) 263 for step in result.steps: 264 yield step
Execute the task, yielding each step as it completes.
76class A2AClient(BaseAgent): 77 """ 78 Proxies ``execute()`` calls to a remote A2A-compliant agent service. 79 80 Uses the Agent2Agent (A2A) protocol — JSON-RPC 2.0 ``tasks/send`` over HTTP. 81 Context (user_assertion, obo_token, locale, etc.) is forwarded as 82 ``message.metadata`` in the JSON-RPC payload. 83 84 This lets any orchestrator (PipelineOrchestrator, SupervisorOrchestrator, etc.) 85 treat a separately-deployed A2A agent service as a first-class agent — no code 86 changes needed in the orchestrator. 87 88 Args: 89 endpoint_url: Base URL of the remote agent service 90 (e.g. ``"http://search-agent:8080"``). 91 ``/rpc`` is appended automatically. 92 timeout: HTTP request timeout in seconds (default: 30). 93 headers: Extra HTTP headers sent with every request (e.g. ``Authorization``). 94 agent_id: Stable identifier used in logging/metrics. Defaults to ``"A2AClient"``. 95 logger: Optional :class:`BasicLogger`. 96 metrics: Optional :class:`BasicMetricsCollector`. 97 98 Example:: 99 100 agent = A2AClient( 101 endpoint_url="http://search-agent-service:8080", 102 agent_id="search_agent", 103 ) 104 result = await agent.execute("Find recent AI papers") 105 106 Discovery:: 107 108 card = await A2AClient.fetch_agent_card("http://search-agent-service:8080") 109 agent = A2AClient( 110 endpoint_url=card["url"], 111 agent_id=card["name"], 112 ) 113 """ 114 115 def __init__( 116 self, 117 endpoint_url: str, 118 timeout: float = 30.0, 119 headers: Optional[Dict[str, str]] = None, 120 behaviors: Optional[List["BaseBehavior"]] = None, 121 agent_id: Optional[str] = None, 122 logger: Optional[BasicLogger] = None, 123 metrics: Optional[BasicMetricsCollector] = None, 124 ) -> None: 125 super().__init__( 126 llm_gateway=None, 127 behaviors=behaviors, 128 agent_id=agent_id or "A2AClient", 129 logger=logger, 130 metrics=metrics, 131 ) 132 self.endpoint_url = endpoint_url.rstrip("/") 133 self.timeout = timeout 134 self._headers = {"Content-Type": "application/json", **(headers or {})} 135 136 # ------------------------------------------------------------------ 137 # BaseAgent implementation 138 # ------------------------------------------------------------------ 139 140 async def execute( 141 self, task: str, context: Optional[Dict[str, Any]] = None 142 ) -> AgentResult: 143 bctx = BehaviorContext(agent_id=self.agent_id, task=task, metadata=context or {}) 144 bctx = await self._apply_behaviors_before(bctx) 145 146 while True: 147 self._log_execution_start(task) 148 try: 149 result = await self._call_remote(task, context or {}) 150 except Exception as exc: 151 fallback = await self._apply_behaviors_on_error(bctx, exc) 152 if fallback is RETRY_SENTINEL: 153 bctx.attempt += 1 154 continue 155 if fallback is not None: 156 return fallback # type: ignore[return-value] 157 raise 158 159 result = await self._apply_behaviors_after(bctx, result) 160 self._log_execution_end(task, result.success, len(result.steps)) 161 return result 162 163 async def stream_execute( 164 self, task: str, context: Optional[Dict[str, Any]] = None 165 ) -> AsyncIterator[AgentStep]: 166 """Executes remotely then yields each step from the response.""" 167 result = await self.execute(task, context) 168 for step in result.steps: 169 yield step 170 171 # ------------------------------------------------------------------ 172 # A2A JSON-RPC transport 173 # ------------------------------------------------------------------ 174 175 async def _call_remote(self, task: str, context: Dict[str, Any]) -> AgentResult: 176 """Send an A2A tasks/send JSON-RPC 2.0 request and return an AgentResult.""" 177 url = f"{self.endpoint_url}/rpc" 178 request_id = str(uuid4()) 179 task_id = str(uuid4()) 180 181 payload = JsonRpcRequest( 182 id=request_id, 183 method="tasks/send", 184 params=A2ATaskParams( 185 id=task_id, 186 message=A2AMessage( 187 parts=[A2APart(type="text", text=task)], 188 # Context (user_assertion, obo_token, locale, etc.) is carried 189 # in message.metadata — A2A agents read it from there. 190 metadata=context, 191 ), 192 ), 193 ).model_dump() 194 195 self._logger.info( 196 "A2AClient tasks/send", 197 url=url, 198 agent_id=self.agent_id, 199 task_id=task_id, 200 ) 201 202 async with httpx.AsyncClient(timeout=self.timeout) as client: 203 try: 204 response = await client.post(url, json=payload, headers=self._headers) 205 except httpx.ConnectError as exc: 206 raise A2AClientError( 207 f"Cannot reach remote agent at {url}: {exc}" 208 ) from exc 209 except httpx.TimeoutException as exc: 210 raise A2AClientError( 211 f"Remote agent timed out after {self.timeout}s: {exc}" 212 ) from exc 213 214 if response.status_code != 200: 215 raise A2AClientError( 216 f"Remote agent returned HTTP {response.status_code}: {response.text[:200]}" 217 ) 218 219 return self._parse_response(response.text) 220 221 def _parse_response(self, body: str) -> AgentResult: 222 """Parse an A2A JSON-RPC tasks/send response into an AgentResult.""" 223 try: 224 envelope = json.loads(body) 225 except json.JSONDecodeError as exc: 226 raise A2AClientError(f"Remote agent returned non-JSON body: {body[:200]}") from exc 227 228 if "error" in envelope: 229 err = JsonRpcErrorResponse.model_validate(envelope) 230 raise A2AClientError( 231 f"A2A JSON-RPC error {err.error.code}: {err.error.message}" 232 ) 233 234 try: 235 response = JsonRpcResponse.model_validate(envelope) 236 except Exception as exc: 237 raise A2AClientError(f"Malformed A2A response: {exc}") from exc 238 239 task = response.result 240 success = task.status.state == "completed" 241 error_msg: Optional[str] = task.status.message if not success else None 242 243 # Extract text output from the first artifact's first text part 244 output = "" 245 for artifact in task.artifacts: 246 for part in artifact.parts: 247 if part.type == "text": 248 output = part.text or "" 249 break 250 if output: 251 break 252 253 # Steps — validate each raw step dict through AgentStepModel 254 steps: List[AgentStep] = [] 255 for raw_step in task.metadata.get("steps", []): 256 s = AgentStepModel.model_validate(raw_step) 257 steps.append( 258 AgentStep( 259 thought=s.thought, 260 action=s.action, 261 action_input=s.action_input, 262 observation=s.observation, 263 metadata=s.metadata, 264 ) 265 ) 266 267 # Propagate all task metadata except "steps" (already parsed above) 268 metadata = {k: v for k, v in task.metadata.items() if k != "steps"} 269 metadata["a2a_task_id"] = task.id 270 271 return AgentResult( 272 output=output, 273 steps=steps, 274 metadata=metadata, 275 success=success, 276 error=error_msg, 277 ) 278 279 # ------------------------------------------------------------------ 280 # A2A discovery helper 281 # ------------------------------------------------------------------ 282 283 @staticmethod 284 async def fetch_agent_card( 285 base_url: str, 286 timeout: float = 10.0, 287 headers: Optional[Dict[str, str]] = None, 288 ) -> AgentCard: 289 """Fetch the A2A Agent Card from ``GET /.well-known/agent.json``. 290 291 Returns an :class:`AgentCard` instance. Standard fields: ``name``, 292 ``description``, ``url``, ``version``, ``skills``. 293 294 Args: 295 base_url: Base URL of the remote agent (e.g. ``"http://agent:8080"``). 296 timeout: HTTP timeout in seconds. 297 headers: Optional extra request headers. 298 299 Raises: 300 A2AClientError: if the endpoint is unreachable or returns non-200. 301 """ 302 url = f"{base_url.rstrip('/')}/.well-known/agent.json" 303 async with httpx.AsyncClient(timeout=timeout) as client: 304 try: 305 response = await client.get(url, headers=headers or {}) 306 except httpx.ConnectError as exc: 307 raise A2AClientError( 308 f"Cannot reach agent card at {url}: {exc}" 309 ) from exc 310 except httpx.TimeoutException as exc: 311 raise A2AClientError( 312 f"Timed out fetching agent card from {url}: {exc}" 313 ) from exc 314 315 if response.status_code != 200: 316 raise A2AClientError( 317 f"GET {url} returned HTTP {response.status_code}: {response.text[:200]}" 318 ) 319 320 try: 321 return AgentCard.model_validate(response.json()) 322 except json.JSONDecodeError as exc: 323 raise A2AClientError( 324 f"Agent card at {url} returned non-JSON: {response.text[:200]}" 325 ) from exc 326 except Exception as exc: 327 raise A2AClientError( 328 f"Agent card at {url} has invalid schema: {exc}" 329 ) from exc
Proxies execute() calls to a remote A2A-compliant agent service.
Uses the Agent2Agent (A2A) protocol — JSON-RPC 2.0 tasks/send over HTTP.
Context (user_assertion, obo_token, locale, etc.) is forwarded as
message.metadata in the JSON-RPC payload.
This lets any orchestrator (PipelineOrchestrator, SupervisorOrchestrator, etc.) treat a separately-deployed A2A agent service as a first-class agent — no code changes needed in the orchestrator.
Args:
endpoint_url: Base URL of the remote agent service
(e.g. "http://search-agent:8080").
/rpc is appended automatically.
timeout: HTTP request timeout in seconds (default: 30).
headers: Extra HTTP headers sent with every request (e.g. Authorization).
agent_id: Stable identifier used in logging/metrics. Defaults to "A2AClient".
logger: Optional BasicLogger.
metrics: Optional BasicMetricsCollector.
Example::
agent = A2AClient(
endpoint_url="http://search-agent-service:8080",
agent_id="search_agent",
)
result = await agent.execute("Find recent AI papers")
Discovery::
card = await A2AClient.fetch_agent_card("http://search-agent-service:8080")
agent = A2AClient(
endpoint_url=card["url"],
agent_id=card["name"],
)
A2AClient( endpoint_url: str, timeout: float = 30.0, headers: Optional[Dict[str, str]] = None, behaviors: Optional[List[gmf_forge_ai_orchestration.BaseBehavior]] = None, agent_id: Optional[str] = None, logger: Optional[gmf_forge_ai_shared_core.observability.BasicLogger] = None, metrics: Optional[gmf_forge_ai_shared_core.observability.BasicMetricsCollector] = None)
115 def __init__( 116 self, 117 endpoint_url: str, 118 timeout: float = 30.0, 119 headers: Optional[Dict[str, str]] = None, 120 behaviors: Optional[List["BaseBehavior"]] = None, 121 agent_id: Optional[str] = None, 122 logger: Optional[BasicLogger] = None, 123 metrics: Optional[BasicMetricsCollector] = None, 124 ) -> None: 125 super().__init__( 126 llm_gateway=None, 127 behaviors=behaviors, 128 agent_id=agent_id or "A2AClient", 129 logger=logger, 130 metrics=metrics, 131 ) 132 self.endpoint_url = endpoint_url.rstrip("/") 133 self.timeout = timeout 134 self._headers = {"Content-Type": "application/json", **(headers or {})}
140 async def execute( 141 self, task: str, context: Optional[Dict[str, Any]] = None 142 ) -> AgentResult: 143 bctx = BehaviorContext(agent_id=self.agent_id, task=task, metadata=context or {}) 144 bctx = await self._apply_behaviors_before(bctx) 145 146 while True: 147 self._log_execution_start(task) 148 try: 149 result = await self._call_remote(task, context or {}) 150 except Exception as exc: 151 fallback = await self._apply_behaviors_on_error(bctx, exc) 152 if fallback is RETRY_SENTINEL: 153 bctx.attempt += 1 154 continue 155 if fallback is not None: 156 return fallback # type: ignore[return-value] 157 raise 158 159 result = await self._apply_behaviors_after(bctx, result) 160 self._log_execution_end(task, result.success, len(result.steps)) 161 return result
Execute the task and return a result.
async def
stream_execute( self, task: str, context: Optional[Dict[str, Any]] = None) -> AsyncIterator[AgentStep]:
163 async def stream_execute( 164 self, task: str, context: Optional[Dict[str, Any]] = None 165 ) -> AsyncIterator[AgentStep]: 166 """Executes remotely then yields each step from the response.""" 167 result = await self.execute(task, context) 168 for step in result.steps: 169 yield step
Executes remotely then yields each step from the response.
@staticmethod
async def
fetch_agent_card( base_url: str, timeout: float = 10.0, headers: Optional[Dict[str, str]] = None) -> gmf_forge_ai_orchestration.protocols.a2a.AgentCard:
283 @staticmethod 284 async def fetch_agent_card( 285 base_url: str, 286 timeout: float = 10.0, 287 headers: Optional[Dict[str, str]] = None, 288 ) -> AgentCard: 289 """Fetch the A2A Agent Card from ``GET /.well-known/agent.json``. 290 291 Returns an :class:`AgentCard` instance. Standard fields: ``name``, 292 ``description``, ``url``, ``version``, ``skills``. 293 294 Args: 295 base_url: Base URL of the remote agent (e.g. ``"http://agent:8080"``). 296 timeout: HTTP timeout in seconds. 297 headers: Optional extra request headers. 298 299 Raises: 300 A2AClientError: if the endpoint is unreachable or returns non-200. 301 """ 302 url = f"{base_url.rstrip('/')}/.well-known/agent.json" 303 async with httpx.AsyncClient(timeout=timeout) as client: 304 try: 305 response = await client.get(url, headers=headers or {}) 306 except httpx.ConnectError as exc: 307 raise A2AClientError( 308 f"Cannot reach agent card at {url}: {exc}" 309 ) from exc 310 except httpx.TimeoutException as exc: 311 raise A2AClientError( 312 f"Timed out fetching agent card from {url}: {exc}" 313 ) from exc 314 315 if response.status_code != 200: 316 raise A2AClientError( 317 f"GET {url} returned HTTP {response.status_code}: {response.text[:200]}" 318 ) 319 320 try: 321 return AgentCard.model_validate(response.json()) 322 except json.JSONDecodeError as exc: 323 raise A2AClientError( 324 f"Agent card at {url} returned non-JSON: {response.text[:200]}" 325 ) from exc 326 except Exception as exc: 327 raise A2AClientError( 328 f"Agent card at {url} has invalid schema: {exc}" 329 ) from exc
Fetch the A2A Agent Card from GET /.well-known/agent.json.
Returns an AgentCard instance. Standard fields: name,
description, url, version, skills.
Args:
base_url: Base URL of the remote agent (e.g. "http://agent:8080").
timeout: HTTP timeout in seconds.
headers: Optional extra request headers.
Raises: A2AClientError: if the endpoint is unreachable or returns non-200.
class
A2AClientError(builtins.RuntimeError):
72class A2AClientError(RuntimeError): 73 """Raised when the remote A2A agent returns an error, is unreachable, or violates the protocol."""
Raised when the remote A2A agent returns an error, is unreachable, or violates the protocol.
class
A2AAdapter:
43class A2AAdapter: 44 """Protocol-only A2A receiver adapter. 45 46 The adapter does not know about tools, checkpoints, routing, or any other 47 orchestration concerns. It only translates between the A2A wire protocol and 48 the host agent's execution callback. 49 """ 50 51 def __init__( 52 self, 53 agent_id: str, 54 description: str, 55 url: str, 56 version: str = "0.1.0", 57 skills: Optional[Sequence[Mapping[str, Any]]] = None, 58 logger: Optional[BasicLogger] = None, 59 ) -> None: 60 self.agent_id = agent_id 61 self.description = description 62 self.url = url.rstrip("/") 63 self.version = version 64 self._skills: List[Dict[str, Any]] = [dict(skill) for skill in (skills or [])] 65 self._logger = logger or BasicLogger(f"gmf_forge_ai.a2a.{self.agent_id}") 66 67 def agent_card(self) -> AgentCard: 68 """Return the A2A Agent Card for ``GET /.well-known/agent.json``.""" 69 return AgentCard( 70 name=self.agent_id, 71 description=self.description, 72 url=self.url, 73 version=self.version, 74 skills=[AgentSkill(**s) for s in self._skills], 75 ) 76 77 def build_jsonrpc_error( 78 self, 79 rpc_id: Optional[str], 80 code: int, 81 message: str, 82 data: Optional[Any] = None, 83 ) -> Dict[str, Any]: 84 """Build a JSON-RPC 2.0 error response.""" 85 return JsonRpcErrorResponse( 86 id=rpc_id, 87 error=JsonRpcError(code=code, message=message, data=data), 88 ).model_dump() 89 90 def parse_tasks_send(self, body: Mapping[str, Any]) -> tuple[Optional[str], str, str, Dict[str, Any]]: 91 """Parse a ``tasks/send`` JSON-RPC request body. 92 93 Returns: 94 A tuple of ``(rpc_id, task_id, task_text, context)``. 95 """ 96 method = str(body.get("method", "")) 97 if method != "tasks/send": 98 raise A2AAdapterError(f"Method not found: {method}") 99 100 params = body.get("params", {}) 101 if not isinstance(params, Mapping): 102 raise A2AAdapterError("Invalid params payload") 103 104 rpc_id = body.get("id") 105 task_id = str(params.get("id") or uuid4()) 106 107 message = params.get("message", {}) 108 if not isinstance(message, Mapping): 109 raise A2AAdapterError("Invalid message payload") 110 111 task_text = self._extract_task_text(message.get("parts", [])) 112 context = message.get("metadata", {}) 113 if not isinstance(context, Mapping): 114 raise A2AAdapterError("Invalid message metadata payload") 115 116 return rpc_id, task_id, task_text, dict(context) 117 118 def build_task_response( 119 self, 120 rpc_id: Optional[str], 121 task_id: str, 122 result: AgentResult, 123 ) -> Dict[str, Any]: 124 """Convert an :class:`AgentResult` into an A2A JSON-RPC success response.""" 125 metadata = dict(result.metadata) 126 metadata["steps"] = [self._serialize_step(step) for step in result.steps] 127 metadata.setdefault("a2a_task_id", task_id) 128 129 artifacts = ( 130 [A2AArtifact(parts=[A2APart(type="text", text=result.output)])] 131 if result.output 132 else [] 133 ) 134 135 return JsonRpcResponse( 136 id=rpc_id, 137 result=A2ATask( 138 id=task_id, 139 status=A2ATaskStatus( 140 state="completed" if result.success else "failed", 141 message=result.error if not result.success else None, 142 ), 143 artifacts=artifacts, 144 metadata=metadata, 145 ), 146 ).model_dump() 147 148 async def handle_rpc( 149 self, 150 body: Mapping[str, Any], 151 execute: Callable[[str, Dict[str, Any]], Awaitable[AgentResult]], 152 ) -> Dict[str, Any]: 153 """Handle an incoming A2A JSON-RPC body and return the response payload.""" 154 rpc_id = body.get("id") if isinstance(body, Mapping) else None 155 156 try: 157 rpc_id, task_id, task_text, context = self.parse_tasks_send(body) 158 except A2AAdapterError as exc: 159 code = -32601 if str(exc).startswith("Method not found") else -32602 160 return self.build_jsonrpc_error(rpc_id if isinstance(rpc_id, str) else None, code, str(exc)) 161 162 self._logger.info( 163 "A2A tasks/send received", 164 agent_id=self.agent_id, 165 task_id=task_id, 166 task_preview=task_text[:80], 167 ) 168 169 try: 170 result = await execute(task_text, context) 171 except Exception as exc: 172 self._logger.error( 173 "Agent execution failed", 174 agent_id=self.agent_id, 175 task_id=task_id, 176 error=str(exc), 177 ) 178 return self.build_jsonrpc_error( 179 rpc_id if isinstance(rpc_id, str) else None, 180 -32603, 181 str(exc), 182 ) 183 184 return self.build_task_response(rpc_id if isinstance(rpc_id, str) else None, task_id, result) 185 186 def _extract_task_text(self, parts: Any) -> str: 187 if not isinstance(parts, Sequence): 188 raise A2AAdapterError("Invalid message parts payload") 189 190 for part in parts: 191 if isinstance(part, Mapping) and part.get("type") == "text": 192 return str(part.get("text", "")) 193 return "" 194 195 def _serialize_step(self, step: AgentStep) -> Dict[str, Any]: 196 return AgentStepModel( 197 thought=step.thought, 198 action=step.action, 199 action_input=step.action_input, 200 observation=step.observation, 201 metadata=step.metadata, 202 ).model_dump()
Protocol-only A2A receiver adapter.
The adapter does not know about tools, checkpoints, routing, or any other orchestration concerns. It only translates between the A2A wire protocol and the host agent's execution callback.
A2AAdapter( agent_id: str, description: str, url: str, version: str = '0.1.0', skills: Optional[Sequence[Mapping[str, Any]]] = None, logger: Optional[gmf_forge_ai_shared_core.observability.BasicLogger] = None)
51 def __init__( 52 self, 53 agent_id: str, 54 description: str, 55 url: str, 56 version: str = "0.1.0", 57 skills: Optional[Sequence[Mapping[str, Any]]] = None, 58 logger: Optional[BasicLogger] = None, 59 ) -> None: 60 self.agent_id = agent_id 61 self.description = description 62 self.url = url.rstrip("/") 63 self.version = version 64 self._skills: List[Dict[str, Any]] = [dict(skill) for skill in (skills or [])] 65 self._logger = logger or BasicLogger(f"gmf_forge_ai.a2a.{self.agent_id}")
67 def agent_card(self) -> AgentCard: 68 """Return the A2A Agent Card for ``GET /.well-known/agent.json``.""" 69 return AgentCard( 70 name=self.agent_id, 71 description=self.description, 72 url=self.url, 73 version=self.version, 74 skills=[AgentSkill(**s) for s in self._skills], 75 )
Return the A2A Agent Card for GET /.well-known/agent.json.
def
build_jsonrpc_error( self, rpc_id: Optional[str], code: int, message: str, data: Optional[Any] = None) -> Dict[str, Any]:
77 def build_jsonrpc_error( 78 self, 79 rpc_id: Optional[str], 80 code: int, 81 message: str, 82 data: Optional[Any] = None, 83 ) -> Dict[str, Any]: 84 """Build a JSON-RPC 2.0 error response.""" 85 return JsonRpcErrorResponse( 86 id=rpc_id, 87 error=JsonRpcError(code=code, message=message, data=data), 88 ).model_dump()
Build a JSON-RPC 2.0 error response.
def
parse_tasks_send( self, body: Mapping[str, typing.Any]) -> tuple[typing.Optional[str], str, str, typing.Dict[str, typing.Any]]:
90 def parse_tasks_send(self, body: Mapping[str, Any]) -> tuple[Optional[str], str, str, Dict[str, Any]]: 91 """Parse a ``tasks/send`` JSON-RPC request body. 92 93 Returns: 94 A tuple of ``(rpc_id, task_id, task_text, context)``. 95 """ 96 method = str(body.get("method", "")) 97 if method != "tasks/send": 98 raise A2AAdapterError(f"Method not found: {method}") 99 100 params = body.get("params", {}) 101 if not isinstance(params, Mapping): 102 raise A2AAdapterError("Invalid params payload") 103 104 rpc_id = body.get("id") 105 task_id = str(params.get("id") or uuid4()) 106 107 message = params.get("message", {}) 108 if not isinstance(message, Mapping): 109 raise A2AAdapterError("Invalid message payload") 110 111 task_text = self._extract_task_text(message.get("parts", [])) 112 context = message.get("metadata", {}) 113 if not isinstance(context, Mapping): 114 raise A2AAdapterError("Invalid message metadata payload") 115 116 return rpc_id, task_id, task_text, dict(context)
Parse a tasks/send JSON-RPC request body.
Returns:
A tuple of (rpc_id, task_id, task_text, context).
def
build_task_response( self, rpc_id: Optional[str], task_id: str, result: AgentResult) -> Dict[str, Any]:
118 def build_task_response( 119 self, 120 rpc_id: Optional[str], 121 task_id: str, 122 result: AgentResult, 123 ) -> Dict[str, Any]: 124 """Convert an :class:`AgentResult` into an A2A JSON-RPC success response.""" 125 metadata = dict(result.metadata) 126 metadata["steps"] = [self._serialize_step(step) for step in result.steps] 127 metadata.setdefault("a2a_task_id", task_id) 128 129 artifacts = ( 130 [A2AArtifact(parts=[A2APart(type="text", text=result.output)])] 131 if result.output 132 else [] 133 ) 134 135 return JsonRpcResponse( 136 id=rpc_id, 137 result=A2ATask( 138 id=task_id, 139 status=A2ATaskStatus( 140 state="completed" if result.success else "failed", 141 message=result.error if not result.success else None, 142 ), 143 artifacts=artifacts, 144 metadata=metadata, 145 ), 146 ).model_dump()
Convert an AgentResult into an A2A JSON-RPC success response.
async def
handle_rpc( self, body: Mapping[str, typing.Any], execute: Callable[[str, typing.Dict[str, typing.Any]], Awaitable[AgentResult]]) -> Dict[str, Any]:
148 async def handle_rpc( 149 self, 150 body: Mapping[str, Any], 151 execute: Callable[[str, Dict[str, Any]], Awaitable[AgentResult]], 152 ) -> Dict[str, Any]: 153 """Handle an incoming A2A JSON-RPC body and return the response payload.""" 154 rpc_id = body.get("id") if isinstance(body, Mapping) else None 155 156 try: 157 rpc_id, task_id, task_text, context = self.parse_tasks_send(body) 158 except A2AAdapterError as exc: 159 code = -32601 if str(exc).startswith("Method not found") else -32602 160 return self.build_jsonrpc_error(rpc_id if isinstance(rpc_id, str) else None, code, str(exc)) 161 162 self._logger.info( 163 "A2A tasks/send received", 164 agent_id=self.agent_id, 165 task_id=task_id, 166 task_preview=task_text[:80], 167 ) 168 169 try: 170 result = await execute(task_text, context) 171 except Exception as exc: 172 self._logger.error( 173 "Agent execution failed", 174 agent_id=self.agent_id, 175 task_id=task_id, 176 error=str(exc), 177 ) 178 return self.build_jsonrpc_error( 179 rpc_id if isinstance(rpc_id, str) else None, 180 -32603, 181 str(exc), 182 ) 183 184 return self.build_task_response(rpc_id if isinstance(rpc_id, str) else None, task_id, result)
Handle an incoming A2A JSON-RPC body and return the response payload.
class
A2AAdapterError(builtins.RuntimeError):
39class A2AAdapterError(RuntimeError): 40 """Raised when an incoming A2A request is malformed or unsupported."""
Raised when an incoming A2A request is malformed or unsupported.