ReAct: reason, act, observe, repeat

"""
ReAct: reason, act, observe, repeat

The loop that sits underneath most working agents. The model does not answer
directly. It alternates between reasoning about what it needs and acting to
get it, feeding each observation back until it can answer.

    Thought -> Action -> Observation -> Thought -> ... -> Answer

What separates ReAct from plain tool-calling is that the reasoning is an
explicit, inspectable field rather than something that happened inside the
model. That is the property worth keeping.

Two things this shape does NOT give you for free, both deliberate here:

  - Nothing stops an action. Every action the model picks runs. Compose with
    the dry-run pattern if any action is irreversible.
  - Nothing steers it. Once started, the loop runs to MAX_STEPS or to an
    answer. You can abort it; you cannot redirect it.

Run: python react_loop.py
"""

import ast
import operator
from typing import Literal

from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel, Field

load_dotenv()

client = OpenAI()

MODEL = "gpt-5.4-mini"
MAX_STEPS = 6

# --------------------------------------------------------------
# Tools
# --------------------------------------------------------------

INVENTORY = {
    "widget": {"stock": 12, "unit_price": 9.99},
    "gadget": {"stock": 0, "unit_price": 24.50},
    "sprocket": {"stock": 340, "unit_price": 1.25},
}


def lookup_item(name: str) -> str:
    item = INVENTORY.get(name.strip().lower())
    if item is None:
        return f"unknown item '{name}'. Known items: {', '.join(INVENTORY)}"
    return f"{name}: stock={item['stock']}, unit_price={item['unit_price']}"


#: The only operations the calculator will perform. Anything outside this map
#: is refused by structure rather than by pattern-matching the input.
_BINARY_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
}
_UNARY_OPS = {ast.UAdd: operator.pos, ast.USub: operator.neg}


def _evaluate_node(node: ast.expr) -> float:
    """Walk a parsed expression, admitting only numbers and the four operators.

    Deliberately not `eval`. A tool an agent can call with model-authored input
    is the last place to hand over an interpreter, and a character allowlist in
    front of `eval` is a filter someone will eventually widen by one character.
    Refusing by structure cannot be widened by accident.
    """
    if isinstance(node, ast.Constant):
        if isinstance(node.value, bool) or not isinstance(node.value, (int, float)):
            raise ValueError(f"not a number: {node.value!r}")
        return float(node.value)
    if isinstance(node, ast.BinOp) and type(node.op) in _BINARY_OPS:
        return _BINARY_OPS[type(node.op)](
            _evaluate_node(node.left), _evaluate_node(node.right)
        )
    if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPS:
        return _UNARY_OPS[type(node.op)](_evaluate_node(node.operand))
    raise ValueError(f"unsupported expression element: {type(node).__name__}")


def calculate(expression: str) -> str:
    """Arithmetic only, over numbers and + - * / with parentheses."""
    if not expression.strip():
        return "refused: empty expression"
    try:
        parsed = ast.parse(expression, mode="eval")
        return str(round(_evaluate_node(parsed.body), 4))
    except ZeroDivisionError:
        return f"could not evaluate '{expression}': division by zero"
    except (SyntaxError, ValueError, TypeError) as exc:
        return f"refused: '{expression}' is not plain arithmetic ({exc})"


TOOLS = {"lookup_item": lookup_item, "calculate": calculate}

TOOL_DOCS = """
lookup_item(name)      -> stock and unit price for one inventory item
calculate(expression)  -> evaluate a plain arithmetic expression
""".strip()


# --------------------------------------------------------------
# One turn of the loop
# --------------------------------------------------------------


class Step(BaseModel):
    thought: str = Field(
        description="What you know so far and what you still need. Always required."
    )
    next: Literal["act", "answer"]
    action: Literal["lookup_item", "calculate"] | None = None
    action_input: str | None = None
    answer: str | None = Field(
        default=None, description="Set only when next == 'answer'."
    )


def think(question: str, transcript: list[str]) -> Step:
    history = "\n".join(transcript) if transcript else "(nothing yet)"
    completion = client.beta.chat.completions.parse(
        model=MODEL,
        messages=[
            {
                "role": "system",
                "content": (
                    "You answer questions by reasoning and using tools, one step "
                    "at a time.\n\nTools:\n" + TOOL_DOCS + "\n\n"
                    "State your reasoning in `thought` every step. Set next='act' "
                    "with an action to gather information, or next='answer' when "
                    "the transcript already contains everything you need. Never "
                    "guess a number you could look up."
                ),
            },
            {
                "role": "user",
                "content": f"Question: {question}\n\nTranscript so far:\n{history}",
            },
        ],
        response_format=Step,
    )
    return completion.choices[0].message.parsed


def run(question: str) -> str:
    transcript: list[str] = []

    for step_number in range(1, MAX_STEPS + 1):
        # Emitted BEFORE the model call, not after it. A step that is still
        # running is visible as "running", which is the whole point.
        print(f"  [{step_number}/{MAX_STEPS}] thinking...")

        step = think(question, transcript)
        print(f"      thought: {step.thought}")
        transcript.append(f"Thought: {step.thought}")

        if step.next == "answer":
            answer = step.answer or "(model chose to answer but gave none)"
            print(f"      answer : {answer}")
            return answer

        if not step.action or step.action not in TOOLS:
            observation = f"no such tool: {step.action}"
        else:
            print(f"      action : {step.action}({step.action_input or ''})")
            observation = TOOLS[step.action](step.action_input or "")

        print(f"      observed: {observation}")
        transcript.append(
            f"Action: {step.action}({step.action_input}) -> {observation}"
        )

    # Budget exhausted is a real outcome, not a crash. Say so plainly and
    # hand back what was learned rather than raising.
    exhausted = (
        f"Stopped after {MAX_STEPS} steps without reaching an answer. "
        f"What was established:\n" + "\n".join(transcript)
    )
    print("      BUDGET EXHAUSTED")
    return exhausted


if __name__ == "__main__":
    for question in [
        "What does it cost to buy every widget we have in stock?",
        "Can I order 5 gadgets right now?",
    ]:
        print(f"\n> {question}")
        run(question)