In the astream_log() method, you cannot use the bind method with RunnableLambda.

[jason1315]

In my project, I need to implement the following logic. Here is a simple:

import asyncio

from langchain_core.runnables import *

from lang_chain.llm.llms import llm

def _test(_dict):
print("value:", _dict)
return _dict

@chain
def my_method(_dict, **keywords):
print(keywords)
return RunnablePassthrough.assign(key=lambda x: keywords.get("i")) | RunnableLambda(_test)

if name == 'main':

loop = asyncio.new_event_loop()

my_list = ["1", "2", "3", " 4", "5"]

head = RunnablePassthrough()
for i in my_list:
    head = head | my_method.bind(i=i)
stream = head.invoke({})
#
# async def __stream(stream1):
#     async for i in stream1:
#         print(i)
#
# loop.run_until_complete(__stream(stream))

When I use the .invoke({}) method, it outputs the following results correctly:

{'i': '1'}
value: {'key': '1'}
{'i': '2'}
value: {'key': '2'}
{'i': '3'}
value: {'key': '3'}
{'i': ' 4'}
value: {'key': ' 4'}
{'i': '5'}
value: {'key': '5'}

But if I use the astream_log({}) method, it throws an error:

File "F:\py3.11\Lib\concurrent\futures\thread.py", line 58, in run
result = self.fn(*self.args, **self.kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: RunnableLambda._atransform.<locals>.func() got an unexpected keyword argument 'i'

Why is it designed like this? Do I need to implement a runnable similar to the model if I want to achieve the above logic?

[dosubot[bot]]

🤖

Hello @jason1315! I'm Dosu, a bot designed to assist while you're waiting for a human maintainer. Whether it's solving bugs, answering questions, or helping you become a contributor, I'm here to make your experience smoother. Let's take a look at your issue together, and I'll provide a full answer shortly.

[jason1315]

How to use the bind method in RunnableLambda?

[dosubot[bot]]

🤖

The error you're encountering is due to the way the bind method is designed to work with RunnableLambda. The bind method is used to bind arguments to the function encapsulated by the RunnableLambda instance. However, the astream_log({}) method is not a function encapsulated by a RunnableLambda instance, but a method of the RunnableLambda class itself.

Here's an example of how to use the bind method correctly:

from langchain_core.runnables import RunnableLambda

def add(x: int, y: int) -> int:
    return x + y

runnable = RunnableLambda(add)
bound_runnable = runnable.bind(y=1)

print(bound_runnable.invoke(2))  # returns 3

In this example, the bind method is used to bind the argument y=1 to the function add. The invoke method is then used to call the function with the bound argument and an additional argument x=2, resulting in the output 3.

Regarding the bind method implementation in LangChain, it's not directly implemented in the RunnableLambda class. Instead, the concept of binding is represented through the RunnableBinding class. This class is used to create a new runnable from an existing one, with some arguments pre-filled (bound).

Here's the relevant code from the RunnableBinding class:

def __init__(
        self,
        *,
        bound: Runnable[Input, Output],
        kwargs: Optional[Mapping[str, Any]] = None,
        config: Optional[RunnableConfig] = None,
        config_factories: Optional[
            List[Callable[[RunnableConfig], RunnableConfig]]
        ] = None,
        custom_input_type: Optional[Union[Type[Input], BaseModel]] = None,
        custom_output_type: Optional[Union[Type[Output], BaseModel]] = None,
        **other_kwargs: Any,
    ) -> None:
        """Create a RunnableBinding from a runnable and kwargs.

The invoke method of the RunnableBinding class then uses these bound arguments when invoking the runnable:

def invoke(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Optional[Any],
    ) -> Output:
        return self.bound.invoke(
            input,
            self._merge_configs(config),
            **{**self.kwargs, **kwargs},
        )

In this method, self.kwargs represents the bound arguments, and they are merged with any additional arguments provided when the method is called.

You can find more details in the LangChain codebase.

Sources

• libs/core/langchain_core/runnables/base.py

---

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.

[jason1315]

So, you mean that I cannot use the bind method under the astream_log method with RunnableLambda, and if I want to use the bind method, I need to implement a class myself?

[1315577677]

I have the same question too.

[eyurtsev]

I created an issue here: #17241 to handle the binding issue.

The suggestion from dosu is not the usual way to work with RunnableLambdas. The first argument is meant to be used as a dict to accommodate multiple arguments, other arguments are used for callbacks and internal debugging things. We might change this behavior to make it easier to work with runnable lambdas, but would need to confirm there are no issues.

Use Runnable Lambdas with an inputs dict to handle multiple inputs:

from langchain_core.runnables import RunnableLambda

def my_func(inputs) -> int:
    if 'x' not in inputs:
        raise KeyError('missing `x`')
    if 'y' not in inputs:
        raise KeyError('missing `y`')
        
    x = inputs['x']
    y = inputs['y']
   return x + y
        

runnable = RunnableLambda(my_func)
print(runnable.invoke({"x": 3, "y": 4}))
print(runnable.invoke({"x": 3, "y": 4}))

And binding would look like this -- but I am guessing that binding is actually is unnecessary to achieve the correct behavior

from langchain_core.runnables import RunnableLambda

def my_func(inputs, **kwargs) -> int:
    if 'x' not in inputs:
        raise KeyError('missing `x`')
    if 'y' not in inputs:
        raise KeyError('missing `y`')
        
    x = inputs['x']
    y = inputs['y']
        
    op = kwargs.get('op', 'add')
    
    if op == 'add':
        return x + y
    elif op == 'mul':
        return x * y
    else:
        return 42

runnable = RunnableLambda(my_func)
print(runnable.bind(op='add').invoke({"x": 3, "y": 4}))
print(runnable.bind(op='mul').invoke({"x": 3, "y": 4}))

[ccurme]

I believe this was addressed here: #17739