Skip to content

Advanced Guide: Stateful Agents (ReAct)

The StatefulOrchestrator is designed to build complex agents that can reason, act, and loop until a task is complete. A classic example is the ReAct (Reason + Act) pattern.

This is a complete, runnable script for a ReAct agent.

from safeagent import *
import json

# --- Setup ---
gov = GovernanceManager()
cfg = Config()
llm = LLMClient(api_key=cfg.api_key)
tool_registry = ToolRegistry(governance_manager=gov, embedding_config={"api_key": cfg.api_key})

@tool_registry.register()
def search_wikipedia(query: str) -> str:
    """Searches for a term on Wikipedia. Use this for general knowledge questions."""
    print(f"TOOL: Searching Wikipedia for '{query}'")
    if "zephyrhills" in query.lower():
        return "Zephyrhills is a city in Pasco County, Florida, known for its bottled water."
    return "No information found for that query."

# --- Agent Nodes ---
def reason_node(state: dict) -> dict:
    """Decides which tool to use or if the task is complete."""
    print("--- STEP: REASONING ---")

    prompt = f"Question: {state['question']}\n"
    prompt += f"Conversation History: {state.get('history', [])}\n\n"
    prompt += "Decide the next action. Should you use a tool, or do you have the final answer? "
    prompt += "If using a tool, respond with JSON: {\"tool_name\": \"tool_to_use\", \"tool_args\": {\"arg\": \"value\"}}. "
    prompt += "Otherwise, respond with the final answer as a plain string."

    response = llm.generate(prompt)

    try:
        decision = json.loads(response['text'])
        print(f"Decision: Call tool '{decision.get('tool_name')}'")
        return {"decision": decision}
    except json.JSONDecodeError:
        print(f"Decision: Final Answer Found")
        return {"decision": {"tool_name": "__end__", "final_answer": response['text']}}

def act_node(state: dict) -> dict:
    """Executes the chosen tool."""
    print("--- STEP: ACTING ---")
    tool_name = state['decision']['tool_name']
    tool_args = state['decision']['tool_args']

    governed_tool = tool_registry.get_governed_tool(tool_name, user_id="agent_123")
    observation = governed_tool(**tool_args)

    print(f"Observation: {observation}")
    new_history = state.get('history', []) + [f"Called tool '{tool_name}' and got this result: '{observation}'"]
    return {"history": new_history}

# --- Define Conditional Edge ---
def should_continue(state: dict) -> str:
    """The routing logic for the graph. It checks the reasoning decision."""
    if state['decision']['tool_name'] == "__end__":
        return "__end__"
    return "act_node"

# --- Build and Run the Graph ---
agent = StatefulOrchestrator(entry_node="reason_node")
agent.add_node("reason_node", reason_node)
agent.add_node("act_node", act_node)
agent.add_edge("act_node", "reason_node") 
agent.add_conditional_edge("reason_node", should_continue)

initial_state = {"question": "What is Zephyrhills known for?"}
gov.start_new_run()
status, final_state = agent.run(initial_state)

print("\n--- AGENT FINISHED ---")
print(f"Status: {status}")
print(f"Final Answer: {final_state.get('decision', {}).get('final_answer')}")