configurable agent tool at runtime

[rere950303]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

pass

Description

How can I make the tool used by the agent to configure the argument in runtime as in lcel?

System Info

pass

[dosubot[bot]]

🤖

Hello again, @rere950303! It's great to see you diving into another interesting challenge with LangChain. How's everything going?

To configure the arguments used by the agent tool in LangChain at runtime, you can modify the args_schema in the create_prompt method of the StructuredChatAgent class. This method is responsible for creating the prompt that is used by the agent, and it includes the arguments for each tool.

Here is the relevant code snippet:

@classmethod
def create_prompt(
    cls,
    tools: Sequence[BaseTool],
    prefix: str = PREFIX,
    suffix: str = SUFFIX,
    human_message_template: str = HUMAN_MESSAGE_TEMPLATE,
    format_instructions: str = FORMAT_INSTRUCTIONS,
    input_variables: Optional[List[str]] = None,
    memory_prompts: Optional[List[BasePromptTemplate]] = None,
) -> BasePromptTemplate:
    tool_strings = []
    for tool in tools:
        args_schema = re.sub("}", "}}", re.sub("{", "{{", str(tool.args)))
        tool_strings.append(f"{tool.name}: {tool.description}, args: {args_schema}")
    ...

In this method, args_schema is created by converting the args attribute of each tool to a string. If you want to configure the arguments at runtime, you can modify this line to use a different method of obtaining the arguments, or modify the args attribute of the tools before they are passed to this method. You can find this code in the StructuredChatAgent class.

The args attribute in the BaseTool class in LangChain is a dictionary that represents the schema of the arguments that the tool accepts. This is determined by the args_schema attribute of the tool, if it is not None. If args_schema is None, then the schema is created from the function associated with the tool. You can find this code in the BaseTool class.

As for the lcel mentioned in your question, it refers to the LangChain Expression Language (LCEL). It is a declarative method to build production-grade programs that harness the power of Language Models (LLMs). The LCEL and LangChain Runnables inherently support synchronous, asynchronous, batch, and streaming operations. This allows servers hosting LCEL based programs to scale better for higher concurrent loads and for creating more responsive User Experiences (UX) by streaming intermediate outputs as they're being generated.

As for configuring arguments at runtime, the LangChain framework provides several utilities for this purpose. The ConfigurableField, ConfigurableFieldSingleOption, and ConfigurableFieldMultiOption classes, imported from langchain.schema.runnable.utils, are likely used to define fields that can be configured at runtime. The RunnableConfig and patch_config from langchain.schema.runnable.config are also likely involved in the runtime configuration of arguments. However, without more specific code or documentation, it's hard to say exactly how these classes and functions are used to configure arguments at runtime. You can find this code in the Runnable module.

I hope this helps! If you have any more questions, feel free to ask.

Sources

• libs/langchain/langchain/agents/structured_chat/base.py
• libs/core/langchain_core/tools.py
• libs/langchain/langchain/schema/runnable/init.py

About Dosu

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[Dalouvid]

I get the idea of using a bot for a first suggestion. But guys, this discourages users from replying to answers.
Please can some human answer this question.

[NourEldin-Osama]

any updates on how to use ConfigurableField with agent?

[maximeperrindev]

Hi all,

To configure a tool behavior at runtime, you can create your own version of a RunnableSerializable class.
Let's implement this with a configurable field name.

from langchain_core.runnables import ConfigurableField, RunnableSerializable
from langchain_core.runnables.config import RunnableConfig
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.prompts import ChatPromptTemplate

class MyCustomRunnable(RunnableSerializable[list[str], str]):
    name: str

    def invoke(self, input: list[str], config: RunnableConfig | None = None) -> str:
        #Do your tool computing using input[0] to get the query parameter and self.name the runtime configured field
        return f"{self.name} - {input[0]}" 

@tool
def my_tool(
    query: str
):
    """
    Tool description
    """

    return (
        MyCustomRunnable()
        .configurable_fields(
            name=ConfigurableField(
                id="name",
                name="name",
                description="Name configured at runtime.",
            )
        )
        .invoke(input=[query])
    )

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a helpful assistant.",
        ),
        ("placeholder", "{chat_history}"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]
)

# Construct the Tools agent
agent = create_tool_calling_agent(llm, [my_tool], prompt)

# Create an agent executor by passing in the agent and tools
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
agent_executor.invoke({"input": "what is LangChain?", "config": {"configurable" : {"name": "MY RUNTIME FIELD"}}})

If you are using Langserve you might need to update the configuration for each request :

async def per_req_config_modifier(config: Dict, request: Request) -> Dict:
    """Modify the config for each request."""
    req = await request.json()
    if "name" in req["config"]["configurable"]:
        config["configurable"]["name] = req["config"]["configurable"]["name"]
    return config

add_routes(
    app,
    agent_executor,
    path="/chat",
    per_req_config_modifier=per_req_config_modifier,
)

Let me know if you have any questions !

[samuelhei]

Hey @maximeperrindev I am trying to run this code (modified a little bit just to run):

from langchain_core.runnables import ConfigurableField, RunnableSerializable
from langchain_core.runnables.config import RunnableConfig
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import tool
from langchain_community.llms import Ollama
from langchain import hub

class MyCustomRunnable(RunnableSerializable[list[str], str]):
    name: str = None

    def invoke(self, input: list[str], config: RunnableConfig | None = None) -> str:
        #Do your tool computing using input[0] to get the query parameter and self.name the runtime configured field
        print(f"{self.name} - {input[0]}" )
        exit()
        return f"{self.name} - {input[0]}" 

@tool
def my_tool(
    query: str
):
    """
    Get information about langchain
    """

    return (
        MyCustomRunnable()
        .configurable_fields(
            name=ConfigurableField(
                id="name",
                name="name",
                description="Name configured at runtime.",
            )
        )
        .invoke(input=[query])
    )

prompt = hub.pull("hwchase17/react")

tools = [my_tool]

llm = Ollama(model="starcoder2:3b")

agent = create_react_agent(llm,tools, prompt)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
agent_executor.invoke({"input": "what is LangChain?", "config": {"configurable" : {"name": "MY RUNTIME FIELD"}}})

And it is returning: None - What is langchain?

Do you know how to fix it?

[maximeperrindev]

Hi @samuelhei,

Still trying to figure out why it is not working without Langserve config...

It seems the configuration propagation for configurable key is not working very well since i'm getting this config in the MyCustomRunnable: {'tags': [], 'metadata': {}, 'callbacks': <langchain_core.callbacks.manager.CallbackManager object at 0x12af869d0>, 'recursion_limit': 25, 'run_id': None}

I'm gonna try to update this discussion if I find anything relevant or a solution (might be a bug)

[samuelhei]

Hi @samuelhei,

Still trying to figure out why it is not working without Langserve config...

It seems the configuration propagation for configurable key is not working very well since i'm getting this config in the MyCustomRunnable: {'tags': [], 'metadata': {}, 'callbacks': <langchain_core.callbacks.manager.CallbackManager object at 0x12af869d0>, 'recursion_limit': 25, 'run_id': None}

I'm gonna try to update this discussion if I find anything relevant or a solution (might be a bug)

Thank you!

[NachoBoe]

Hi! Any updates on this?

I believe that the "None" you are recieving is because the agent_executer doesn´t propagate the config to the tools, therefore it uses the default name: str = None defined in your class

The work around I found is to use a wrapper on the agent executor, that do propagate the tools:

`
class CustomAgentExecutor(Runnable):
"""A custom runnable that will be used by the agent executor."""

def __init__(self, agent,tools, **kwargs):
    """Initialize the runnable."""
    super().__init__(**kwargs)
    self.agent = agent
    self.tools = tools

def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
    """Will not be used."""
    raise NotImplementedError()

@property
def config_specs(self) -> List[ConfigurableFieldSpec]:
    return self.agent.config_specs

async def astream(
    self,
    input: Input,
    config: Optional[RunnableConfig] = None,
    **kwargs: Optional[Any],
) -> AsyncIterator[Output]:
    """Stream the agent's output."""
    configurable = cast(Dict[str, Any], config.pop("configurable", {}))
    configured_tools = []
    if configurable:
        for tool in self.tools:
            configured_tools.append(tool.with_config({"configurable":configurable}))
    executor = AgentExecutor(
        agent=self.agent,
        tools=configured_tools,
    ).with_config({"run_name": "executor"})

    async for output in executor.astream(input, config=config, **kwargs):
        yield output

`
This is similar to this example: https://github.com/langchain-ai/langserve/blob/main/examples/configurable_agent_executor/server.py but propagating the config to the tools instead of the agent.

One extra problem I found is that using with_config on the tools change their type from StructuredTool to RunnableBinding, which doesnt have _run implemented. Therefore the agent executor fails the validation of the tools. The work around for this is to define the funcion with_config on the tool definition:

`

class EmptyInput(BaseModel):
pass

class CustomSearchTool(BaseTool):
name = "user_id_getter"
description = "Retrieves user id"
args_schema: Type[BaseModel] = EmptyInput
config = {}

def _run(
    self,run_manager: Optional[CallbackManagerForToolRun] = None
) -> str:
    """Use the tool."""
    return self.config

async def _arun(
    self,  run_manager: Optional[AsyncCallbackManagerForToolRun] = None
) -> str:
    """Use the tool asynchronously."""
    return self.config

def with_config(self,config):
    self.config = config
    return self

`

However, It would be nice that tools natively accepted configurations without changing their type to runnable, and that agentExecutor propagated configuration to the tools. Or maybe I am missing something

[samuelhei]

@rere950303 I did a very similar workaround, I created a child class of Tool modifiing both run and arun methods adding this part:

    child_config = patch_config(
                config,
                callbacks=run_manager.get_child(),
            )
    tool_args, tool_kwargs = self._to_args_and_kwargs(parsed_input)
    tool_kwargs['config'] = child_config #Explicitly passing a config arg to the tool

and in the function definition I added the **kargs to get the config variable

def search(query, **kargs):
    user_id = kargs.get('config',{'user_id': 'NOT_SET'}).get('user_id','NOT_SET')

[yvesloy]

Do you the full code of a working workaround? I want to get this to run!