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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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
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) 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 = getattr(tool, "args")
        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 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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:
            parsed_tool_args = {}
            tool = self.tools[pred.next_tool_name]
            for k, v in pred.next_tool_args.items():
                if hasattr(tool, "arg_types") and k in tool.arg_types:
                    arg_type = tool.arg_types[k]
                    if isinstance((origin := get_origin(arg_type) or arg_type), type) and issubclass(
                        origin, BaseModel
                    ):
                        parsed_tool_args[k] = arg_type.model_validate(v)
                        continue
                parsed_tool_args[k] = v
            trajectory[f"observation_{idx}"] = self.tools[pred.next_tool_name](**parsed_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)