""" Multi-step agent executor implementing the ReAct pattern. ReAct (Reason + Act) interleaves thinking and tool use: the model reasons about what to do, calls a tool, observes the result, then decides whether to call another tool or give a final answer. The loop has hard guardrails for a small model: * *max_steps* (default 5) prevents infinite loops. * If the model never produces ````, the last response is used. * Tool errors are fed back as observations so the model can recover. * Repeated identical tool calls break the loop early. The executor is pure Python with no Streamlit dependency. It takes a ``GenerationEngine`` and ``ToolRegistry``, and returns an ``AgentResult``. """ from __future__ import annotations from typing import Iterator from src.agents.parser import coerce_param, parse_tool_call from src.agents.registry import ToolRegistry from src.agents.schemas import AgentResult, AgentStep, ToolCall # Optional guardrails import — available when src.guardrails is installed. try: from src.guardrails.middleware import GuardrailsMiddleware except ImportError: # pragma: no cover GuardrailsMiddleware = None # type: ignore[assignment, misc] # --------------------------------------------------------------------------- # System prompt template # --------------------------------------------------------------------------- _SYSTEM_TEMPLATE = """\ You are a helpful assistant with access to tools. ## How to use a tool Thought: your reasoning about what to do tool_name value ## How to give a final answer Thought: your reasoning your final answer here ## Rules - Use exactly ONE tool per response. - Always wrap your final answer in tags. - If you can answer without tools, use immediately. - When the user asks about documents, files, policies, reports, revenue, or project-specific information, ALWAYS use retrieve_documents first — do not answer from your own knowledge or use web_search. - Only use web_search for current events, live data, or topics clearly outside the uploaded documents. - If unsure whether to use retrieve_documents or web_search, prefer retrieve_documents. ## Examples User: What does the document say about pricing? Assistant: Thought: The user is asking about document content, so I should search the uploaded documents. retrieve_documents pricing User: What models does the project support? Assistant: Thought: This is about uploaded project documentation, so I should search the documents. retrieve_documents supported models User: What is the average salary? Assistant: Thought: I need to query the data table for the average salary. query_data SELECT AVG(salary) FROM employees User: What is the latest news about Python? Assistant: Thought: This is about current events, so I need to search the web. web_search latest Python news User: Hello! Assistant: Thought: This is a greeting, no tools needed. Hello! How can I help you today? {tool_descriptions}""" class AgentExecutor: """ReAct agent loop with tool calling.""" def __init__( self, engine, registry: ToolRegistry, max_steps: int = 5, temperature: float = 0.3, max_tokens_per_step: int = 300, guardrails: GuardrailsMiddleware | None = None, extra_instructions: str = "", ) -> None: self.engine = engine self.registry = registry self.max_steps = max_steps self.temperature = temperature self.max_tokens_per_step = max_tokens_per_step self.guardrails = guardrails # Optional domain-specific rules appended to the system prompt # (e.g. financial period-grounding), kept out of the generic template. self.extra_instructions = extra_instructions # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def run( self, user_query: str, chat_history: list[dict] | None = None, ) -> AgentResult: """Execute the agent loop and return the final result.""" # Scan user input. if self.guardrails: input_scan = self.guardrails.scan_input(user_query) if input_scan.blocked: return AgentResult( answer="I can't process that request — it was flagged by safety filters.", steps=[], num_steps=0, finished=True, guardrail_flags=[f.__dict__ for f in input_scan.flags], ) steps: list[AgentStep] = [] system_prompt = self._build_system_prompt() prev_call: tuple[str, str] | None = None # (tool_name, args_key) for repeat detection for _ in range(self.max_steps): messages = self._build_step_messages( system_prompt, user_query, steps, chat_history, ) raw_output = self.engine.generate_answer( messages, max_tokens=self.max_tokens_per_step, temperature=self.temperature, ) thought, tool_call, final_answer = parse_tool_call(raw_output) # Final answer path. if final_answer is not None: step = AgentStep(thought=thought) steps.append(step) result = AgentResult( answer=final_answer, steps=steps, num_steps=len(steps), finished=True, ) return self._scan_answer(result) # Tool call path. assert tool_call is not None call_key = (tool_call.tool_name, str(sorted(tool_call.arguments.items()))) # Detect repeated identical tool call. if call_key == prev_call: step = AgentStep( thought=thought, tool_call=tool_call, error="Repeated tool call — stopping.", ) steps.append(step) break prev_call = call_key observation = self._execute_tool(tool_call) step = AgentStep( thought=thought, tool_call=tool_call, observation=observation, ) steps.append(step) # Exhausted max_steps — use last observation or raw output as answer. fallback = steps[-1].observation if steps else "" result = AgentResult( answer=fallback or "I was unable to determine an answer.", steps=steps, num_steps=len(steps), finished=False, ) return self._scan_answer(result) def run_stream( self, user_query: str, chat_history: list[dict] | None = None, ) -> Iterator[tuple[int, str, str]]: """Yield ``(step_index, event_type, content)`` tuples for streaming UI. *event_type* is one of: ``"thought"``, ``"tool_call"``, ``"observation"``, ``"answer"``, ``"guardrail"``. """ # Scan user input. if self.guardrails: input_scan = self.guardrails.scan_input(user_query) if input_scan.blocked: yield 0, "guardrail", "; ".join(f.description for f in input_scan.flags) yield 0, "answer", "I can't process that request — it was flagged by safety filters." return steps: list[AgentStep] = [] system_prompt = self._build_system_prompt() prev_call: tuple[str, str] | None = None for step_idx in range(self.max_steps): messages = self._build_step_messages( system_prompt, user_query, steps, chat_history, ) raw_output = self.engine.generate_answer( messages, max_tokens=self.max_tokens_per_step, temperature=self.temperature, ) thought, tool_call, final_answer = parse_tool_call(raw_output) if thought: yield step_idx, "thought", thought # Final answer. if final_answer is not None: steps.append(AgentStep(thought=thought)) if self.guardrails: out_scan = self.guardrails.scan_output(final_answer) if out_scan.flags: yield step_idx, "guardrail", "; ".join(f.description for f in out_scan.flags) if out_scan.blocked: yield step_idx, "answer", "The response was blocked by safety filters." return yield step_idx, "answer", final_answer return # Tool call. assert tool_call is not None call_key = (tool_call.tool_name, str(sorted(tool_call.arguments.items()))) call_desc = "{}({})".format( tool_call.tool_name, ", ".join("{}={}".format(k, v) for k, v in tool_call.arguments.items()), ) yield step_idx, "tool_call", call_desc if call_key == prev_call: steps.append(AgentStep( thought=thought, tool_call=tool_call, error="Repeated tool call — stopping.", )) break prev_call = call_key observation = self._execute_tool(tool_call) steps.append(AgentStep( thought=thought, tool_call=tool_call, observation=observation, )) yield step_idx, "observation", observation # Exhausted steps — yield whatever we have. fallback = steps[-1].observation if steps else "I was unable to determine an answer." yield len(steps) - 1, "answer", fallback # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _scan_answer(self, result: AgentResult) -> AgentResult: """Run guardrail checks on the final answer, if configured.""" if not self.guardrails: return result out_scan = self.guardrails.scan_output(result.answer) if out_scan.flags: result.guardrail_flags = [ {"check_name": f.check_name, "severity": f.severity.value, "description": f.description} for f in out_scan.flags ] if out_scan.blocked: result.answer = "The response was blocked by safety filters." return result def _build_system_prompt(self) -> str: prompt = _SYSTEM_TEMPLATE.format( tool_descriptions=self.registry.format_tool_descriptions(), ) if self.extra_instructions: prompt += "\n\n## Additional rules\n" + self.extra_instructions return prompt def _build_step_messages( self, system_prompt: str, user_query: str, steps: list[AgentStep], chat_history: list[dict] | None, ) -> list[dict]: messages: list[dict] = [{"role": "system", "content": system_prompt}] # Include recent chat history (last 4 turns for context window budget). if chat_history: messages.extend(chat_history[-4:]) # Append /no_think for Qwen3 to suppress blocks — they waste # the token budget on internal reasoning that duplicates our Thought: line. # Other architectures (Gemma3, Gemma4) don't use this control token. arch = getattr(self.engine, "arch", "qwen3") suffix = " /no_think" if arch == "qwen3" else "" messages.append({"role": "user", "content": user_query + suffix}) # Encode previous steps as assistant/user turn pairs. for step in steps: # Reconstruct the assistant's raw output. raw = step.tool_call.raw_text if step.tool_call and step.tool_call.raw_text else "" if not raw and step.thought: raw = "Thought: {}".format(step.thought) if raw: messages.append({"role": "assistant", "content": raw}) if step.observation: messages.append({ "role": "user", "content": "Observation: {}".format(step.observation), }) return messages def _execute_tool(self, tool_call: ToolCall) -> str: """Look up and execute the tool, returning the observation string.""" tool = self.registry.get(tool_call.tool_name) if tool is None: return "Error: unknown tool '{}'. Available tools: {}".format( tool_call.tool_name, ", ".join(t.name for t in self.registry.list_tools()), ) if tool.execute is None: return "Error: tool '{}' has no execute function.".format(tool_call.tool_name) # Coerce parameter types. kwargs: dict = {} param_types = {p.name: p.type for p in tool.params} for key, value in tool_call.arguments.items(): expected = param_types.get(key, "str") kwargs[key] = coerce_param(value, expected) try: return tool.execute(**kwargs) except Exception as e: return "Error executing tool '{}': {}".format(tool_call.tool_name, e)