workflow as a tool for multi-agent chtbot

[debarun-ghosh17]

I have separate workflows, example advanced-text-to-sql, I want to use them as tools for a multi-agent chatbot or in a react agent later down the line. Is there a way to achieve this?

[debarun-ghosh17]

@dosu-bot how can we use a workflow as a queryenginetool or a functiontool?

[dosubot[bot]]

To wrap a workflow as a tool for use in a multi-agent chatbot or a ReAct agent using LlamaIndex, you can create a Workflow by defining steps with the @step decorator. Each step handles specific event types and emits new events. Here's a detailed example of how you can set up a workflow, incorporating stateful memory and tool invocation:

from llama_index.core.workflow import (
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
)
from llama_index.llms.openai import OpenAI
from llama_index.core.llms import ChatMessage
from llama_index.core.tools import ToolSelection, ToolOutput
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.agent.react import ReActChatFormatter, ReActOutputParser
from llama_index.core.tools.types import BaseTool

class PrepEvent(Event):
    pass

class InputEvent(Event):
    input: list[ChatMessage]

class ToolCallEvent(Event):
    tool_calls: list[ToolSelection]

class FunctionOutputEvent(Event):
    output: ToolOutput

class ReActAgent(Workflow):
    def __init__(self, llm=None, tools=None, extra_context=None, **kwargs):
        super().__init__(**kwargs)
        self.tools = tools or []
        self.llm = llm or OpenAI()
        self.memory = ChatMemoryBuffer.from_defaults(llm=llm)
        self.formatter = ReActChatFormatter(context=extra_context or "")
        self.output_parser = ReActOutputParser()
        self.sources = []

    @step
    async def new_user_msg(self, ctx, ev: StartEvent) -> PrepEvent:
        self.sources = []
        user_input = ev.input
        user_msg = ChatMessage(role="user", content=user_input)
        self.memory.put(user_msg)
        await ctx.set("current_reasoning", [])
        return PrepEvent()

    @step
    async def prepare_chat_history(self, ctx, ev: PrepEvent) -> InputEvent:
        chat_history = self.memory.get()
        current_reasoning = await ctx.get("current_reasoning", default=[])
        llm_input = self.formatter.format(self.tools, chat_history, current_reasoning=current_reasoning)
        return InputEvent(input=llm_input)

    @step
    async def handle_llm_input(self, ctx, ev: InputEvent) -> ToolCallEvent | StopEvent:
        chat_history = ev.input
        response = await self.llm.achat(chat_history)
        try:
            reasoning_step = self.output_parser.parse(response.message.content)
            (await ctx.get("current_reasoning", default=[])).append(reasoning_step)
            if reasoning_step.is_done:
                self.memory.put(ChatMessage(role="assistant", content=reasoning_step.response))
                return StopEvent(result={"response": reasoning_step.response, "sources": [*self.sources], "reasoning": await ctx.get("current_reasoning", default=[])})
            elif isinstance(reasoning_step, ActionReasoningStep):
                tool_name = reasoning_step.action
                tool_args = reasoning_step.action_input
                return ToolCallEvent(tool_calls=[ToolSelection(tool_id="fake", tool_name=tool_name, tool_kwargs=tool_args)])
        except Exception as e:
            (await ctx.get("current_reasoning", default=[])).append(ObservationReasoningStep(observation=f"There was an error in parsing my reasoning: {e}"))
        return PrepEvent()

    @step
    async def handle_tool_calls(self, ctx, ev: ToolCallEvent) -> PrepEvent:
        tool_calls = ev.tool_calls
        tools_by_name = {tool.metadata.get_name(): tool for tool in self.tools}
        for tool_call in tool_calls:
            tool = tools_by_name.get(tool_call.tool_name)
            if not tool:
                (await ctx.get("current_reasoning", default=[])).append(ObservationReasoningStep(observation=f"Tool {tool_call.tool_name} does not exist"))
                continue
            try:
                tool_output = tool(**tool_call.tool_kwargs)
                self.sources.append(tool_output)
                (await ctx.get("current_reasoning", default=[])).append(ObservationReasoningStep(observation=tool_output.content))
            except Exception as e:
                (await ctx.get("current_reasoning", default=[])).append(ObservationReasoningStep(observation=f"Error calling tool {tool.metadata.get_name()}: {e}"))
        return PrepEvent()

# Running the workflow
agent = ReActAgent(llm=OpenAI(model="gpt-4o-mini"), tools=[...], timeout=120, verbose=True)
result = await agent.run(input="Hello!")
print(result["response"])

This setup allows you to integrate workflows as tools within a multi-agent system, enabling complex interactions and data processing capabilities ^[1][2].

To continue talking to Dosu, mention @dosu.

[debarun-ghosh17]

@dosu the above example is creating a reactagent in workflow. I want an existing workflow I have coded to be used as a tool for a react agent

[dosubot[bot]]

To use an existing workflow as a tool for a ReAct agent, you need to define your workflow as a tool and integrate it into the ReAct agent's setup. Here's how you can do it:

1. Define Your Workflow as a Tool: Wrap your existing workflow in a function that can be called by the ReAct agent. This function should take inputs and return outputs in a way that the ReAct agent can understand.

2. Create a FunctionTool: Use the FunctionTool.from_defaults() method to create a tool from your function. This method allows you to define a function as a tool that the ReAct agent can call.

3. Integrate the Tool into the ReAct Agent: When initializing the ReActAgent, pass your tool (created in the previous step) in the tools parameter.

Here's an example setup:

from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI

# Define your existing workflow as a function
def my_workflow_tool(input_data):
    # Your existing workflow logic here
    # Process input_data and return the result
    return processed_result

# Create a FunctionTool from your workflow function
my_tool = FunctionTool.from_defaults(my_workflow_tool)

# Initialize the ReActAgent with your tool
agent = ReActAgent(
    llm=OpenAI(model="gpt-4o-mini"), tools=[my_tool], timeout=120, verbose=True
)

# Run the agent with an input
ret = await agent.run(input="Your input here")

This setup allows your existing workflow to be used as a tool within the ReAct agent, enabling the agent to call your workflow as part of its decision-making process ^[1].

To continue talking to Dosu, mention @dosu.

[debarun-ghosh17]

@dosu the implementation is going into a loop like 'Running step new_user_msg
Step new_user_msg produced event PrepEvent
Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event PrepEvent
Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event PrepEvent'
Is is because workflows use await, but we are not awaiting the result while enclosing the workflow in a functiontool

[dosubot[bot]]

To prevent your workflow from going into a loop when using it as a FunctionTool in a ReAct agent, you can set a timeout for the workflow execution. This ensures that the workflow does not run indefinitely. For example, you can set a timeout of 120 seconds for the ReActAgent:

agent = ReActAgent(
    llm=OpenAI(model="gpt-4o-mini"), tools=[my_tool], timeout=120, verbose=True
)

By setting this timeout, the workflow will automatically stop if it exceeds the specified duration, preventing it from looping indefinitely ^[1].

To continue talking to Dosu, mention @dosu.

[markns]

Hey @debarun-ghosh17 did you get anywhere with this? I’m also trying to figure out how to integrate workflows and chat engines, and wondering if the chat engine should be embedded/reimplemented in the workflow, or the workflow should be wrapped as a tool for the chat engine.