Runnable that assigns key-value pairs to Dict[str, Any] inputs. The `RunnableAssign` class takes input dictionaries and, through a `RunnableParallel` instance, applies transformations, then combines these with the original data, introducing new key-value pairs based on the mapper's
| 329 | |
| 330 | |
| 331 | class RunnableAssign(RunnableSerializable[Dict[str, Any], Dict[str, Any]]): |
| 332 | """Runnable that assigns key-value pairs to Dict[str, Any] inputs. |
| 333 | |
| 334 | The `RunnableAssign` class takes input dictionaries and, through a |
| 335 | `RunnableParallel` instance, applies transformations, then combines |
| 336 | these with the original data, introducing new key-value pairs based |
| 337 | on the mapper's logic. |
| 338 | |
| 339 | Examples: |
| 340 | .. code-block:: python |
| 341 | |
| 342 | # This is a RunnableAssign |
| 343 | from typing import Dict |
| 344 | from langchain_core.runnables.passthrough import ( |
| 345 | RunnableAssign, |
| 346 | RunnableParallel, |
| 347 | ) |
| 348 | from langchain_core.runnables.base import RunnableLambda |
| 349 | |
| 350 | def add_ten(x: Dict[str, int]) -> Dict[str, int]: |
| 351 | return {"added": x["input"] + 10} |
| 352 | |
| 353 | mapper = RunnableParallel( |
| 354 | {"add_step": RunnableLambda(add_ten),} |
| 355 | ) |
| 356 | |
| 357 | runnable_assign = RunnableAssign(mapper) |
| 358 | |
| 359 | # Synchronous example |
| 360 | runnable_assign.invoke({"input": 5}) |
| 361 | # returns {'input': 5, 'add_step': {'added': 15}} |
| 362 | |
| 363 | # Asynchronous example |
| 364 | await runnable_assign.ainvoke({"input": 5}) |
| 365 | # returns {'input': 5, 'add_step': {'added': 15}} |
| 366 | """ |
| 367 | |
| 368 | mapper: RunnableParallel[Dict[str, Any]] |
| 369 | |
| 370 | def __init__(self, mapper: RunnableParallel[Dict[str, Any]], **kwargs: Any) -> None: |
| 371 | super().__init__(mapper=mapper, **kwargs) # type: ignore[call-arg] |
| 372 | |
| 373 | @classmethod |
| 374 | def is_lc_serializable(cls) -> bool: |
| 375 | return True |
| 376 | |
| 377 | @classmethod |
| 378 | def get_lc_namespace(cls) -> List[str]: |
| 379 | """Get the namespace of the langchain object.""" |
| 380 | return ["langchain", "schema", "runnable"] |
| 381 | |
| 382 | def get_name( |
| 383 | self, suffix: Optional[str] = None, *, name: Optional[str] = None |
| 384 | ) -> str: |
| 385 | name = ( |
| 386 | name |
| 387 | or self.name |
| 388 | or f"RunnableAssign<{','.join(self.mapper.steps__.keys())}>" |