Skip to content

dspy.ReAct

dspy.ReAct(signature, tools: list[Callable], max_iters=5)

Bases: Module

tools is either a list of functions, callable classes, or dspy.Tool instances.

Source code in dspy/predict/react.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(self, signature, tools: list[Callable], max_iters=5):
    """
    `tools` is either a list of functions, callable classes, or `dspy.Tool` instances.
    """

    self.signature = signature = ensure_signature(signature)
    self.max_iters = max_iters

    tools = [t if isinstance(t, Tool) or hasattr(t, "input_variable") else Tool(t) for t in tools]
    tools = {tool.name: tool for tool in tools}

    inputs = ", ".join([f"`{k}`" for k in signature.input_fields.keys()])
    outputs = ", ".join([f"`{k}`" for k in signature.output_fields.keys()])
    instr = [f"{signature.instructions}\n"] if signature.instructions else []

    instr.extend(
        [
            f"You will be given {inputs} and your goal is to finish with {outputs}.\n",
            "To do this, you will interleave Thought, Tool Name, and Tool Args, and receive a resulting Observation.\n",
            "Thought can reason about the current situation, and Tool Name can be the following types:\n",
        ]
    )

    finish_desc = (
        f"Signals that the final outputs, i.e. {outputs}, are now available and marks the task as complete."
    )
    finish_args = {}  # k: v.annotation for k, v in signature.output_fields.items()}
    tools["finish"] = Tool(func=lambda **kwargs: "Completed.", name="finish", desc=finish_desc, args=finish_args)

    for idx, tool in enumerate(tools.values()):
        args = tool.args if hasattr(tool, "args") else str({tool.input_variable: str})
        desc = (f", whose description is <desc>{tool.desc}</desc>." if tool.desc else ".").replace("\n", "  ")
        desc += f" It takes arguments {args} in JSON format."
        instr.append(f"({idx+1}) {tool.name}{desc}")

    react_signature = (
        dspy.Signature({**signature.input_fields}, "\n".join(instr))
        .append("trajectory", dspy.InputField(), type_=str)
        .append("next_thought", dspy.OutputField(), type_=str)
        .append("next_tool_name", dspy.OutputField(), type_=Literal[tuple(tools.keys())])
        .append("next_tool_args", dspy.OutputField(), type_=dict[str, Any])
    )

    fallback_signature = dspy.Signature(
        {**signature.input_fields, **signature.output_fields}, signature.instructions
    ).append("trajectory", dspy.InputField(), type_=str)

    self.tools = tools
    self.react = dspy.Predict(react_signature)
    self.extract = dspy.ChainOfThought(fallback_signature)

Functions

forward(**input_args)

Source code in dspy/predict/react.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def forward(self, **input_args):
    def format(trajectory: dict[str, Any], last_iteration: bool):
        adapter = dspy.settings.adapter or dspy.ChatAdapter()
        trajectory_signature = dspy.Signature(f"{', '.join(trajectory.keys())} -> x")
        return adapter.format_fields(trajectory_signature, trajectory, role="user")

    trajectory = {}
    for idx in range(self.max_iters):
        pred = self.react(**input_args, trajectory=format(trajectory, last_iteration=(idx == self.max_iters - 1)))

        trajectory[f"thought_{idx}"] = pred.next_thought
        trajectory[f"tool_name_{idx}"] = pred.next_tool_name
        trajectory[f"tool_args_{idx}"] = pred.next_tool_args

        try:
            trajectory[f"observation_{idx}"] = self.tools[pred.next_tool_name](**pred.next_tool_args)
        except Exception as e:
            trajectory[f"observation_{idx}"] = f"Failed to execute: {e}"

        if pred.next_tool_name == "finish":
            break

    extract = self.extract(**input_args, trajectory=format(trajectory, last_iteration=False))
    return dspy.Prediction(trajectory=trajectory, **extract)