Replies: 6 comments 17 replies
-
|
To signal to LangChain that a tool has already fulfilled its role and it can stop processing, you can use the Here is an example of how you can yield an async def _aiter_next_step(
self,
name_to_tool_map: Dict[str, BaseTool],
color_mapping: Dict[str, str],
inputs: Dict[str, str],
intermediate_steps: List[Tuple[AgentAction, str]],
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
) -> AsyncIterator[Union[AgentFinish, AgentAction, AgentStep]]:
"""Take a single step in the thought-action-observation loop.
Override this to take control of how the agent makes and acts on choices.
"""
try:
intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)
# Call the LLM to see what to do.
output = await self.agent.aplan(
intermediate_steps,
callbacks=run_manager.get_child() if run_manager else None,
**inputs,
)
except OutputParserException as e:
# Handle parsing errors
# ...
return
# If the tool chosen is the finishing tool, then we end and return.
if isinstance(output, AgentFinish):
yield output
return
actions: List[AgentAction]
if isinstance(output, AgentAction):
actions = [output]
else:
actions = output
for agent_action in actions:
yield agent_action
# Use asyncio.gather to run multiple tool.arun() calls concurrently
result = await asyncio.gather(
*[
self._aperform_agent_action(
name_to_tool_map, color_mapping, agent_action, run_manager
)
for agent_action in actions
],
)
# Yield each result as it becomes available
for chunk in result:
yield chunkIn this example, if the Additionally, in the def on_agent_finish(
self, finish: AgentFinish, color: Optional[str] = None, **kwargs: Any
) -> None:
if self._current_thought is not None:
self._current_thought.complete(
self._thought_labeler.get_final_agent_thought_label()
)
self._current_thought = NoneThis method updates the current thought to indicate completion and ensures that LangChain stops further processing [1][2]. |
Beta Was this translation helpful? Give feedback.
-
|
Hello Dosu!
Thank you very much for your quick response!
Unfortunately, I was not able to understand the answer.
I am not using a Streamlit application. I am running an application on a
FastAPI server. My application streams to a client via WSS.
In this case, how do I implement this solution?
Thank you once again.
Em qui., 27 de jun. de 2024 às 20:06, dosubot[bot] ***@***.***>
escreveu:
… To signal to LangChain that a tool has already fulfilled its role and it
can stop processing, you can use the AgentFinish class. When the tool's
response is sent via streaming through a WebSocket callback and does not
return anything, you can yield an AgentFinish instance to indicate that
the processing should stop.
Here is an example of how you can yield an AgentFinish instance:
async def _aiter_next_step(
self,
name_to_tool_map: Dict[str, BaseTool],
color_mapping: Dict[str, str],
inputs: Dict[str, str],
intermediate_steps: List[Tuple[AgentAction, str]],
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
) -> AsyncIterator[Union[AgentFinish, AgentAction, AgentStep]]:
"""Take a single step in the thought-action-observation loop. Override this to take control of how the agent makes and acts on choices. """
try:
intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)
# Call the LLM to see what to do.
output = await self.agent.aplan(
intermediate_steps,
callbacks=run_manager.get_child() if run_manager else None,
**inputs,
)
except OutputParserException as e:
# Handle parsing errors
# ...
return
# If the tool chosen is the finishing tool, then we end and return.
if isinstance(output, AgentFinish):
yield output
return
actions: List[AgentAction]
if isinstance(output, AgentAction):
actions = [output]
else:
actions = output
for agent_action in actions:
yield agent_action
# Use asyncio.gather to run multiple tool.arun() calls concurrently
result = await asyncio.gather(
*[
self._aperform_agent_action(
name_to_tool_map, color_mapping, agent_action, run_manager
)
for agent_action in actions
],
)
# Yield each result as it becomes available
for chunk in result:
yield chunk
In this example, if the output is an instance of AgentFinish, it is
yielded to indicate that the processing should stop. This approach ensures
that LangChain knows the tool has completed its task and can stop further
processing.
Additionally, in the StreamlitCallbackHandler class, the on_agent_finish
method is used to handle the completion of an agent's task. This method can
be used to signal the end of processing:
def on_agent_finish(
self, finish: AgentFinish, color: Optional[str] = None, **kwargs: Any
) -> None:
if self._current_thought is not None:
self._current_thought.complete(
self._thought_labeler.get_final_agent_thought_label()
)
self._current_thought = None
This method updates the current thought to indicate completion and ensures
that LangChain stops further processing [1]
<https://github.com/langchain-ai/langchain/blob/master/libs/langchain/langchain/agents/agent.py>
[2]
<https://github.com/langchain-ai/langchain/blob/master/libs/community/langchain_community/callbacks/streamlit/streamlit_callback_handler.py>
.
*To continue talking to Dosu <https://dosu.dev>, mention @dosu
<https://github.com/dosu>.*
—
Reply to this email directly, view it on GitHub
<#23612 (comment)>,
or unsubscribe
<https://github.com/notifications/unsubscribe-auth/AAGFVVRPBF7KXEL73UNZUHLZJSLH5AVCNFSM6AAAAABKAZUOYOVHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM4TQOJZGA4TE>
.
You are receiving this because you authored the thread.Message ID:
***@***.***
com>
|
Beta Was this translation helpful? Give feedback.
-
|
Hello @dosu! Thank you very much for your quick response! Unfortunately, I was not able to understand the answer. I am not using a Streamlit application. I am running an application on a FastAPI server. My application streams to a client via WSS. In this case, how do I implement this solution? Thank you once again. |
Beta Was this translation helpful? Give feedback.
-
|
To signal to LangChain that a tool has already fulfilled its role and can stop processing in a FastAPI server environment when the tool's response is sent via streaming through a WebSocket callback, you can use the Here is the relevant code snippet: def on_tool_end(
self,
output: Any,
color: Optional[str] = None,
observation_prefix: Optional[str] = None,
llm_prefix: Optional[str] = None,
**kwargs: Any,
) -> None:
output = str(output)
self._require_current_thought().on_tool_end(
output, color, observation_prefix, llm_prefix, **kwargs
)
self._complete_current_thought()This method converts the tool's output to a string, calls the Additionally, LangChain's events = []
async for event in model.astream_events("hello", version="v2"):
events.append(event)This approach allows you to handle streaming events effectively, ensuring that the tool's completion is properly signaled and managed within the LangChain framework [2]. |
Beta Was this translation helpful? Give feedback.
-
|
@dosu, thank you for your help. My problem persists, and I think it is due to the following behavior: LangChain calls my tool. I believe I need to find a way to signal the LangChain mechanism to consider that the tool executed correctly. As I explained, my application is multi-agent running as a FASTAPI service. Thank you! Here you are my tool source code: ` @tool ` |
Beta Was this translation helpful? Give feedback.
-
|
To ensure LangChain recognizes that your tool has executed correctly and should not be called again, you need to explicitly signal the completion of the tool's task. Since your tool returns the result via streaming through a WebSocket and does not return anything in the function return, LangChain interprets this as if the tool did not execute correctly. You can achieve this by using the
from langchain.agents import AgentFinish
@tool
async def tool_responde_perguntas_e_executa_comandos(pergunta_completa: str):
"""
Útil para responder perguntas, esclarecer dúvidas e executar comandos.
Args:
pergunta (str): Pergunta completa do usuário
"""
chat_agent = get_custom_data("chat_agent")
websocket = chat_agent.websocket
chat = ChatRespondePergunta(chat_agent.app_keys, chat_agent.app_params, websocket)
answer = await chat.chat_responde_pergunta(pergunta_completa)
if websocket is None:
return answer
else:
# Signal that the tool has completed its task
return AgentFinish(
return_values={"output": "ATENÇÃO: Essa ferramenta não retorna resposta. A resposta é enviada diretamente ao usuário de outra forma. Considere a pergunta do humano como já respondida."},
log="Tool execution completed."
)
Make sure your agent is set up to handle the By returning an This approach ensures that LangChain recognizes the tool's completion and stops further processing, addressing the issue of repeated calls [1][2]. |
Beta Was this translation helpful? Give feedback.
-
|
Hello @dosu! Thank you very much! Could you please explain how I can make my agent handle the return from the AgentFinish instance and stop execution? I create my agent like this: self.llm_with_tools = self.llm.bind_tools(self.tools)
agent = create_tool_calling_agent(llm=self.llm_with_tools, tools=self.tools, prompt=myprompt)
resultado = await agent_executor.ainvoke(context, return_only_outputs=True)
agent_executor = AgentExecutor(
agent=agent,
tools=self.tools,
memory=memory,
max_iterations=35,
max_execution_time=90,
early_stopping_method='force',
verbose=True,
handle_parsing_errors=True,
return_intermediate_steps=False,
callbacks=chat.callbacks
).with_config({"run_name": "Agent"})Because by doing just this, it keeps calling the tool in a loop: @tool
async def tool_responde_perguntas(pergunta_completa: str):
"""
This tool is useful for answering questions posed by the user.
"""
#chat_agent = get_custom_data("chat_agent")
websocket = chat_agent.websocket
chat = ChatRespondePergunta(chat_agent.app_keys, chat_agent.app_params, websocket)
answer = await chat.chat_responde_pergunta(pergunta_completa)
if websocket is None:
return answer
else:
return AgentFinish(
return_values={"status": "success"},
log="Tool execution completed."
)Thank you very much! |
Beta Was this translation helpful? Give feedback.
-
|
To make your agent handle the return from the
def _call(
self,
inputs: Dict[str, str],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""Run text through and get agent response."""
# Construct a mapping of tool name to tool for easy lookup
name_to_tool_map = {tool.name: tool for tool in self.tools}
# We construct a mapping from each tool to a color, used for logging.
color_mapping = get_color_mapping(
[tool.name for tool in self.tools], excluded_colors=["green", "red"]
)
intermediate_steps: List[Tuple[AgentAction, str]] = []
# Let's start tracking the number of iterations and time elapsed
iterations = 0
time_elapsed = 0.0
start_time = time.time()
# We now enter the agent loop (until it returns something).
while self._should_continue(iterations, time_elapsed):
next_step_output = self._take_next_step(
name_to_tool_map,
color_mapping,
inputs,
intermediate_steps,
run_manager=run_manager,
)
if isinstance(next_step_output, AgentFinish):
run_manager.on_agent_finish(next_step_output)
return self._return(
next_step_output, intermediate_steps, run_manager=run_manager
)
intermediate_steps.extend(next_step_output)
if len(next_step_output) == 1:
next_step_action = next_step_output[0]
# See if tool should return directly
tool_return = self._get_tool_return(next_step_action)
if tool_return is not None:
return self._return(
tool_return, intermediate_steps, run_manager=run_manager
)
iterations += 1
time_elapsed = time.time() - start_time
output = self.agent.return_stopped_response(
self.early_stopping_method, intermediate_steps, **inputs
)
return self._return(output, intermediate_steps, run_manager=run_manager)
async def _acall(
self,
inputs: Dict[str, str],
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""Run text through and get agent response."""
# Construct a mapping of tool name to tool for easy lookup
name_to_tool_map = {tool.name: tool for tool in self.tools}
# We construct a mapping from each tool to a color, used for logging.
color_mapping = get_color_mapping(
[tool.name for tool in self.tools], excluded_colors=["green", "red"]
)
intermediate_steps: List[Tuple[AgentAction, str]] = []
# Let's start tracking the number of iterations and time elapsed
iterations = 0
time_elapsed = 0.0
start_time = time.time()
# We now enter the agent loop (until it returns something).
while self._should_continue(iterations, time_elapsed):
next_step_output = await self._atake_next_step(
name_to_tool_map,
color_mapping,
inputs,
intermediate_steps,
run_manager=run_manager,
)
if isinstance(next_step_output, AgentFinish):
await run_manager.on_agent_finish(next_step_output)
return await self._areturn(
next_step_output, intermediate_steps, run_manager=run_manager
)
intermediate_steps.extend(next_step_output)
if len(next_step_output) == 1:
next_step_action = next_step_output[0]
# See if tool should return directly
tool_return = self._get_tool_return(next_step_action)
if tool_return is not None:
return await self._areturn(
tool_return, intermediate_steps, run_manager=run_manager
)
iterations += 1
time_elapsed = time.time() - start_time
output = self.agent.return_stopped_response(
self.early_stopping_method, intermediate_steps, **inputs
)
return await self._areturn(output, intermediate_steps, run_manager=run_manager)These methods ensure that when an |
Beta Was this translation helpful? Give feedback.
-
|
Hi @dosu! Thank you again! I don't know if I understood correctly, but will I have to change the source code of LangChain files? How do I do this? Isn't there a way to do this through a callback? Thank you! |
Beta Was this translation helpful? Give feedback.
-
|
You do not have to change the source code of LangChain files. You can achieve this through a callback. LangChain provides a mechanism to handle events through callback handlers, which can be used without modifying the source code. Here is how you can use a callback to signal the completion of the tool's task:
from langchain.callbacks import BaseCallbackHandler
from langchain.agents import AgentFinish
class CustomCallbackHandler(BaseCallbackHandler):
def on_tool_end(self, output, **kwargs):
if isinstance(output, AgentFinish):
print("Tool execution completed.")
# Handle the completion logic here
from langchain.agents import create_tool_calling_agent, AgentExecutor
# Define your tools and prompt
tools = [tool_responde_perguntas]
myprompt = "Your prompt here"
# Create the agent
llm_with_tools = llm.bind_tools(tools)
agent = create_tool_calling_agent(llm=llm_with_tools, tools=tools, prompt=myprompt)
# Create the custom callback handler
custom_callback = CustomCallbackHandler()
# Create the agent executor with the custom callback
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
max_iterations=35,
max_execution_time=90,
early_stopping_method='force',
verbose=True,
handle_parsing_errors=True,
return_intermediate_steps=False,
callbacks=[custom_callback] # Add the custom callback here
).with_config({"run_name": "Agent"})
# Invoke the agent
resultado = await agent_executor.ainvoke(context, return_only_outputs=True)
from langchain.agents import AgentFinish
@tool
async def tool_responde_perguntas(pergunta_completa: str):
"""
This tool is useful for answering questions posed by the user.
"""
websocket = chat_agent.websocket
chat = ChatRespondePergunta(chat_agent.app_keys, chat_agent.app_params, websocket)
answer = await chat.chat_responde_pergunta(pergunta_completa)
if websocket is None:
return answer
else:
return AgentFinish(
return_values={"status": "success"},
log="Tool execution completed."
)By using a custom callback handler, you can handle the completion of the tool's task without modifying the LangChain source code. This approach ensures that LangChain recognizes the tool's completion and stops further processing [1][2]. |
Beta Was this translation helpful? Give feedback.
-
|
Hey @dosu, this didn't work. The tool continues to be called indefinitely. It doesn't call the Thank you! |
Beta Was this translation helpful? Give feedback.
-
|
To handle the Here is how you can implement the necessary functions in the
async def get_graph(self, inputs, intermediate_steps, run_manager=None):
# Prepare inputs
inputs = {**inputs, **{"intermediate_steps": intermediate_steps}}
final_output = None
if self.stream_runnable:
async for chunk in self.runnable.astream(inputs, config={"callbacks": callbacks}):
if final_output is None:
final_output = chunk
else:
final_output += chunk
else:
final_output = await self.runnable.ainvoke(inputs, config={"callbacks": callbacks})
# Check if the output is an AgentFinish instance
if isinstance(final_output, AgentFinish):
yield final_output
return
# Process actions if not finished
actions = [final_output] if isinstance(final_output, AgentAction) else final_output
for agent_action in actions:
yield agent_action
yield await self._aperform_agent_action(name_to_tool_map, color_mapping, agent_action, run_manager)
Here is the updated def get_graph(self):
websocket = self.websocket
app_keys = self.app_keys
app_params = self.app_params
chat = LLMService(app_keys, app_params, websocket)
self.streaming = False
chat.update_streaming(self.streaming)
self.memory = AsyncSqliteSaver.from_conn_string(":memory:")
self.tool_executor = ToolExecutor(self.tools)
self.llm = chat.cria_llm()
self.llm_with_tools = self.llm.bind_tools(self.tools)
graph_builder = StateGraph(State)
graph_builder.add_node("agent", self.run_agent)
graph_builder.add_node("tools", self.tool_node)
graph_builder.add_edge("agent", "tools")
graph_builder.set_entry_point("agent")
graph = graph_builder.compile(checkpointer=self.memory)
try:
#display(Image(graph.get_graph().draw_mermaid_png()))
pass
except Exception:
# This requires some extra dependencies and is optional
pass
return graph
async def run_agent(self, state: State, config: RunnableConfig):
messages = self.filter_messages(state["messages"])
human_message = messages[-1]
prompt = self.format_prompt_context()
response = await self.llm_with_tools.ainvoke([SystemMessage(content=prompt)], config)
# Check if the response is an AgentFinish instance
if isinstance(response, AgentFinish):
return {"messages": [response]}
return {"messages": [response]}By implementing these changes, you ensure that the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu, is it possible to avoid .astream? Is it possible to use things like: |
Beta Was this translation helpful? Give feedback.
-
|
Yes, you can avoid using
def should_continue(data):
# If the agent outcome is an AgentFinish, then we return `end` string
if isinstance(data['agent_outcome'], AgentFinish):
return "end"
# Otherwise, an AgentAction is returned
# Here we return `continue` string
else:
return "continue"
Here is the updated def get_graph(self):
websocket = self.websocket
app_keys = self.app_keys
app_params = self.app_params
chat = LLMService(app_keys, app_params, websocket)
self.streaming = False
chat.update_streaming(self.streaming)
self.memory = AsyncSqliteSaver.from_conn_string(":memory:")
self.tool_executor = ToolExecutor(self.tools)
self.llm = chat.cria_llm()
self.llm_with_tools = self.llm.bind_tools(self.tools)
graph_builder = StateGraph(State)
# Add nodes
graph_builder.add_node("agent", self.run_agent)
graph_builder.add_node("tools", self.tool_node)
# Add edges
graph_builder.add_edge("agent", "tools")
graph_builder.add_conditional_edges(
"tools",
should_continue,
{
"continue": "agent",
"end": END,
},
)
# Set entry point
graph_builder.set_entry_point("agent")
# Compile the graph
graph = graph_builder.compile(checkpointer=self.memory)
try:
# Optionally display the graph
# display(Image(graph.get_graph().draw_mermaid_png()))
pass
except Exception:
# This requires some extra dependencies and is optional
pass
return graph
async def run_agent(self, state: State, config: RunnableConfig):
messages = self.filter_messages(state["messages"])
human_message = messages[-1]
prompt = self.format_prompt_context()
response = await self.llm_with_tools.ainvoke([SystemMessage(content=prompt)], config)
# Include the agent outcome in the state
state['agent_outcome'] = response
return {"messages": [response]}By implementing these changes, you ensure that the |
Beta Was this translation helpful? Give feedback.
-
|
data[ 'agent_outcome' ] is empty here: Maybe "self.tool_node" is not saving the tool response in it? |
Beta Was this translation helpful? Give feedback.
-
|
To ensure that Here is how you can achieve this:
Here is the relevant code: def _process_next_step_output(
self,
next_step_output: Union[AgentFinish, List[Tuple[AgentAction, str]]],
run_manager: CallbackManagerForChainRun,
) -> AddableDict:
"""
Process the output of the next step,
handling AgentFinish and tool return cases.
"""
logger.debug("Processing output of Agent loop step")
if isinstance(next_step_output, AgentFinish):
logger.debug(
"Hit AgentFinish: _return -> on_chain_end -> run final output logic"
)
return self._return(next_step_output, run_manager=run_manager)
self.intermediate_steps.extend(next_step_output)
logger.debug("Updated intermediate_steps with step output")
# Check for tool return
if len(next_step_output) == 1:
next_step_action = next_step_output[0]
tool_return = self.agent_executor._get_tool_return(next_step_action)
if tool_return is not None:
return self._return(tool_return, run_manager=run_manager)
return AddableDict(intermediate_step=next_step_output)
def _return(
self, output: AgentFinish, run_manager: CallbackManagerForChainRun
) -> AddableDict:
"""
Return the final output of the iterator.
"""
returned_output = self.agent_executor._return(
output, self.intermediate_steps, run_manager=run_manager
)
returned_output["messages"] = output.messages
run_manager.on_chain_end(returned_output)
return self.make_final_outputs(returned_output, run_manager)In the
In the
Additionally, you can refer to the # If the tool chosen is the finishing tool, then we end and return.
if isinstance(output, AgentFinish):
yield output
return
actions: List[AgentAction]
if isinstance(output, AgentAction):
actions = [output]
else:
actions = output
for agent_action in actions:
yield agent_action
for agent_action in actions:
yield self._perform_agent_action(
name_to_tool_map, color_mapping, agent_action, run_manager
)This ensures that |
Beta Was this translation helpful? Give feedback.
-
Checked other resources
Commit to Help
Example Code
Description
I am developing a multi-agent ChatBot using create_tool_calling_agent.
My application has dozens of tools, and one of them has a special behavior that I am having difficulty dealing with, which is the reason for my question: it returns the response via streaming through a regular WebSocket callback.
Since the tool's response is sent via streaming, it does not return anything, and thus LangChain understands that the tool needs to be called again, leading it to be executed repeatedly until the maximum limit of max_iterations is reached.
My question is, how can I signal to LangChain that the tool has already fulfilled its role, everything went well, and it can consider the mission accomplished and stop processing?
As this is a conceptual question, I have not included any example code.
Thank you!
System Info
As this is a conceptual question, I have not included anything here.
Beta Was this translation helpful? Give feedback.
All reactions