ReAct Prompting: Reason and Act

Thought: I need to find the mechanism of warfarin before checking its interactions.
Action: search["warfarin mechanism of action"]
Observation: Warfarin inhibits VKOR enzyme, blocking vitamin K recycling...
Thought: Now I know the mechanism. I need to check which drugs inhibit the same pathway.
Action: search["drugs that interact with warfarin via VKOR"]
Observation: NSAIDs, azole antifungals, macrolide antibiotics...
Thought: I have enough information to answer comprehensively.
Final Answer: [complete answer]

from openai import OpenAI
import json

client = OpenAI()

REACT_SYSTEM_PROMPT = """You are a clinical pharmacology assistant with access to tools.
When answering questions, use the following format:

Thought: [What you're thinking about / what you need to find out]
Action: tool_name[input]
Observation: [Result from the tool — this will be filled in by the system]
... (repeat Thought/Action/Observation as needed)
Thought: I have enough information to answer.
Final Answer: [Your complete answer]

Available tools:
- search[query]: Search medical literature for clinical information
- drug_database[drug_name]: Look up drug properties, interactions, and contraindications
- calculate[expression]: Perform numerical calculations (e.g., creatinine clearance)
- patient_record[patient_id]: Retrieve patient's current medications and lab values

Always think before acting. Use tools when you need specific information you don't have."""

def parse_react_output(text: str) -> dict:
    """Parse ReAct-formatted output into structured components."""
    lines = text.strip().split("\n")
    result = {"thoughts": [], "actions": [], "observations": [], "final_answer": None}

    for line in lines:
        if line.startswith("Thought:"):
            result["thoughts"].append(line[8:].strip())
        elif line.startswith("Action:"):
            action_text = line[7:].strip()
            # Parse tool_name[input] format
            if "[" in action_text and action_text.endswith("]"):
                tool = action_text[:action_text.index("[")]
                arg = action_text[action_text.index("[")+1:-1]
                result["actions"].append({"tool": tool, "arg": arg})
        elif line.startswith("Final Answer:"):
            result["final_answer"] = line[13:].strip()

    return result

# Define actual tool implementations
def search(query: str) -> str:
    """Simulate a medical literature search."""
    # In production: call PubMed API, UpToDate, or similar
    knowledge_base = {
        "warfarin mechanism": "Warfarin inhibits VKOR (vitamin K epoxide reductase), preventing regeneration of vitamin K hydroquinone needed for clotting factor synthesis.",
        "warfarin clarithromycin interaction": "Clarithromycin inhibits CYP3A4 and CYP2C9, the primary enzymes metabolizing warfarin (especially S-warfarin via 2C9). This increases warfarin plasma levels and INR, raising bleeding risk.",
        "INR monitoring warfarin": "INR should be checked within 3-5 days of starting an interacting antibiotic, and again after completing the course.",
    }
    for key, value in knowledge_base.items():
        if any(word in query.lower() for word in key.split()):
            return value
    return "No specific information found. Consider checking prescribing reference."

def drug_database(drug_name: str) -> str:
    db = {
        "warfarin": "Anticoagulant; narrow therapeutic index; metabolized by CYP2C9 (S-warfarin) and CYP3A4 (R-warfarin); target INR 2.0-3.0 for most indications.",
        "clarithromycin": "Macrolide antibiotic; strong CYP3A4 inhibitor; moderate CYP2C9 inhibitor; commonly used for respiratory infections.",
    }
    return db.get(drug_name.lower(), f"Drug '{drug_name}' not found in database.")

def calculate(expression: str) -> str:
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Calculation error: {e}"

TOOLS = {
    "search": search,
    "drug_database": drug_database,
    "calculate": calculate,
}

def react_agent(question: str, max_steps: int = 6) -> str:
    """Run the ReAct loop until Final Answer or max_steps."""
    messages = [
        {"role": "system", "content": REACT_SYSTEM_PROMPT},
        {"role": "user", "content": question},
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            temperature=0,
            stop=["Observation:"],  # Stop before filling in the observation
        )

        assistant_output = response.choices[0].message.content
        print(f"\n[Step {step + 1}]\n{assistant_output}")

        # Check if we have a final answer
        if "Final Answer:" in assistant_output:
            final = assistant_output.split("Final Answer:")[-1].strip()
            return final

        # Parse the action
        parsed = parse_react_output(assistant_output)
        if not parsed["actions"]:
            # No action found — prompt the model to continue
            messages.append({"role": "assistant", "content": assistant_output})
            messages.append({"role": "user", "content": "Continue reasoning. Use a tool or provide your Final Answer."})
            continue

        # Execute the tool
        last_action = parsed["actions"][-1]
        tool_name = last_action["tool"]
        tool_arg = last_action["arg"]

        if tool_name in TOOLS:
            observation = TOOLS[tool_name](tool_arg)
        else:
            observation = f"Unknown tool: {tool_name}"

        print(f"Observation: {observation}")

        # Add to conversation: assistant output + observation
        full_turn = assistant_output + f"\nObservation: {observation}"
        messages.append({"role": "assistant", "content": full_turn})

    return "Maximum steps reached without a final answer."

# Run
question = "A patient on warfarin starts clarithromycin 500mg twice daily for 10 days. Their INR was 2.4 yesterday. What should we do?"
answer = react_agent(question)
print(f"\n=== FINAL ANSWER ===\n{answer}")

from pydantic import BaseModel
from typing import Optional

class ReActStep(BaseModel):
    thought: str
    action_tool: Optional[str] = None
    action_input: Optional[str] = None
    observation: Optional[str] = None

class ReActTrace(BaseModel):
    steps: list[ReActStep]
    final_answer: Optional[str] = None
    total_tool_calls: int = 0

    def to_markdown(self) -> str:
        lines = []
        for i, step in enumerate(self.steps, 1):
            lines.append(f"**Step {i}**")
            lines.append(f"- Thought: {step.thought}")
            if step.action_tool:
                lines.append(f"- Action: `{step.action_tool}[{step.action_input}]`")
            if step.observation:
                lines.append(f"- Observation: {step.observation}")
        if self.final_answer:
            lines.append(f"\n**Final Answer:** {self.final_answer}")
        return "\n".join(lines)

def detect_loop(actions: list[dict], window: int = 3) -> bool:
    """Detect if the agent is repeating the same action."""
    if len(actions) < window:
        return False
    recent = actions[-window:]
    return all(a == recent[0] for a in recent)

tools = [
    {
        "type": "function",
        "function": {
            "name": "search",
            "description": "Search medical literature for clinical information about drugs, interactions, or treatments",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Medical search query"}
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "drug_database",
            "description": "Look up drug properties, dosing, interactions, and contraindications",
            "parameters": {
                "type": "object",
                "properties": {
                    "drug_name": {"type": "string", "description": "Name of the drug to look up"}
                },
                "required": ["drug_name"],
            },
        },
    },
]

def react_with_function_calling(question: str, max_iterations: int = 5) -> str:
    messages = [{"role": "user", "content": question}]

    for _ in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",
        )

        msg = response.choices[0].message
        messages.append(msg)

        if response.choices[0].finish_reason == "stop":
            return msg.content

        # Execute all tool calls
        for tool_call in (msg.tool_calls or []):
            args = json.loads(tool_call.function.arguments)
            result = TOOLS[tool_call.function.name](**args)
            messages.append({
                "role": "tool",
                "content": result,
                "tool_call_id": tool_call.id,
            })

    return "Reached maximum iterations."