D_icR ,SrSSKJr SSKrSSKrSSKrSSKrSSKrSSKrSSK J r J r SSK J r JrJrJrJrJrJrJr SSKJrJr SSKJr SSKJr SS KJr SS KJr SS KJ r J!r!J"r"J#r#J$r$J%r%J&r&J'r'J(r(J)r) SS K*J+r+J,r,J-r-J.r. SS K/J0r0 SSK1J2r2 SSK3J4r4J5r5 SSK6J7r7J8r8J9r9 SSK:J;r;Jr>J?r?J@r@JArAJBrBJCrCJDrDJErEJFrF SSKGJHrHJIrIJJrJJKrKJLrLJMrMJNrNJOrOJPrPJQrQJRrRJSrSJTrTJUrUJVrVJWrWJXrXJYrY SSKZJ[r[ SSK\J]r]J^r^ SSK_J`r`Jara SSKbJcrcJdrd SSKeJfrfJgrg SSKhJiri SSKjJkrk \ (a@SSK3JlrlJmrm SSKnJoro SSKpJqrr SSKsJtrt SSKuJvrv SSKwJxrx SS KyJzrz SS!K_J{r{J|r| SS"KbJ}r} SS#K~Jr \%"S$5rS%r"S&S'\ \"\L\M45r"S(S)\7\\L\M45rSJS*jrSJS+jrS%r"S,S-\\L\M45r"S.S/\\L\\\!445r\r"S0S1\\L\M45r"S2S3\\L\M45r"S4S5\\\L\\M45r"S6S7\\L\M45r"S8S9\\L\M45r"S:S;\\L\M45r"S<S=\$\L\M45r"S>S?\$\L\M45r"S@SA\$\L\M45r"SBSC\$\L\M45r\\L\M4\\L/\M4-\\L/\\M4-\\\L/\\M4-\\\L/\\M4-\\L\M4-\\L\M4-\\L\M4-\\L\M4-\\\!4-rSKSDjr\)SLSEj5r\)SMSFj5r\)SNSGj5r\)SOSHj5rSPSIjrg)Qz+Base classes and utilities for `Runnable`s.) annotationsN)ABCabstractmethod)AsyncGenerator AsyncIterator AwaitableCallable CoroutineIteratorMappingSequence)FIRST_COMPLETEDwait)wraps)tee) itemgetter) GenericAlias) TYPE_CHECKINGAnyGenericLiteralProtocolTypeVarcastget_argsget_type_hintsoverload) BaseModel ConfigDictField RootModel)override)beta_decorator)AsyncCallbackManagerCallbackManager) SerializableSerializedConstructorSerializedNotImplemented) RunnableConfigacall_func_with_variable_argscall_func_with_variable_args ensure_config%get_async_callback_manager_for_configget_callback_manager_for_configget_config_listget_executor_for_config merge_configs patch_configrun_in_executorset_config_context) AddableDictAnyConfigurableFieldConfigurableFieldConfigurableFieldSpecInputOutputaccepts_configaccepts_run_managercoro_with_context gated_corogather_with_concurrency get_function_first_arg_dict_keysget_function_nonlocalsget_lambda_sourceget_unique_config_specsindent_lines_after_firstis_async_callableis_async_generator)_StreamingCallbackHandler)!_astream_events_implementation_v1!_astream_events_implementation_v2LogStreamCallbackHandler_astream_log_implementation)AsyncRootListenersTracerRootListenersTracer)aclosingatee)safetee)create_model_v2)AsyncCallbackManagerForChainRunCallbackManagerForChainRunBasePromptTemplateRunnableWithFallbacksGraph)ExponentialJitterParams) StreamEvent)BaseTool)RunLog RunLogPatch) AsyncListener)RunOtherc l\rSrSr%SrS\S'SMSS.SNSjjjr\SOSj5r\SPS j5r \SQS j5r SMSRS jjr SMSSS jjr \SQS j5r SMSRSjjrSMSSSjjr\STSj5rSS.SUSjjrSS.SVSjjrSMSWSjjrSMSXSjjrSYSjrSZSjrSS.S[SjjrS\SjrS]Sjr\SMS^Sjj5rSMS^SjjrSMSS.S_Sjjjr\SMSS.S`S jjj5r\SMSaS!jj5rSMSS.SbS"jjjrSMSS.S_S#jjjr \SMSS.ScS$jjj5r!\SMSdS%jj5r!SMSS.SeS&jjjr!SMSfS'jjr"SMSgS(jjr#\SMS)S)SSSSSSS*.ShS+jjj5r$\SMS)SSSSSSS,.SiS-jjj5r$SMS)S)SSSSSSS*.SjS.jjjr$SMS/SSSSSSS0.SkS1jjjr%SMSlS2jjr&SMSmS3jjr'SnS4jr(SMSoS5jjr)SSSS6.SpS7jjr*SSSS6.SqS8jjr+SSS9.SrS:jjr,\-4S)SS;S<.SsS=jjr.StS>jr/\-4SS?.SuS@jjr0SvSwSAjjr1SvSxSBjjr2SMSSSC.SySDjjjr3SMSSSC.SzSEjjjr4SMS{SFjjr5SMS|SGjjr6\7Rp"SHSI9SMSSSSJ.S}SKjjj5r9SLr:g)~Runnable|aA unit of work that can be invoked, batched, streamed, transformed and composed. Key Methods =========== - `invoke`/`ainvoke`: Transforms a single input into an output. - `batch`/`abatch`: Efficiently transforms multiple inputs into outputs. - `stream`/`astream`: Streams output from a single input as it's produced. - `astream_log`: Streams output and selected intermediate results from an input. Built-in optimizations: - **Batch**: By default, batch runs invoke() in parallel using a thread pool executor. Override to optimize batching. - **Async**: Methods with `'a'` suffix are asynchronous. By default, they execute the sync counterpart using asyncio's thread pool. Override for native async. All methods accept an optional config argument, which can be used to configure execution, add tags and metadata for tracing and debugging etc. Runnables expose schematic information about their input, output and config via the `input_schema` property, the `output_schema` property and `config_schema` method. Composition =========== Runnable objects can be composed together to create chains in a declarative way. Any chain constructed this way will automatically have sync, async, batch, and streaming support. The main composition primitives are `RunnableSequence` and `RunnableParallel`. **`RunnableSequence`** invokes a series of runnables sequentially, with one Runnable's output serving as the next's input. Construct using the `|` operator or by passing a list of runnables to `RunnableSequence`. **`RunnableParallel`** invokes runnables concurrently, providing the same input to each. Construct it using a dict literal within a sequence or by passing a dict to `RunnableParallel`. For example, ```python from langchain_core.runnables import RunnableLambda # A RunnableSequence constructed using the `|` operator sequence = RunnableLambda(lambda x: x + 1) | RunnableLambda(lambda x: x * 2) sequence.invoke(1) # 4 sequence.batch([1, 2, 3]) # [4, 6, 8] # A sequence that contains a RunnableParallel constructed using a dict literal sequence = RunnableLambda(lambda x: x + 1) | { "mul_2": RunnableLambda(lambda x: x * 2), "mul_5": RunnableLambda(lambda x: x * 5), } sequence.invoke(1) # {'mul_2': 4, 'mul_5': 10} ``` Standard Methods ================ All `Runnable`s expose additional methods that can be used to modify their behavior (e.g., add a retry policy, add lifecycle listeners, make them configurable, etc.). These methods will work on any `Runnable`, including `Runnable` chains constructed by composing other `Runnable`s. See the individual methods for details. For example, ```python from langchain_core.runnables import RunnableLambda import random def add_one(x: int) -> int: return x + 1 def buggy_double(y: int) -> int: """Buggy code that will fail 70% of the time""" if random.random() > 0.3: print('This code failed, and will probably be retried!') # noqa: T201 raise ValueError('Triggered buggy code') return y * 2 sequence = ( RunnableLambda(add_one) | RunnableLambda(buggy_double).with_retry( # Retry on failure stop_after_attempt=10, wait_exponential_jitter=False ) ) print(sequence.input_schema.model_json_schema()) # Show inferred input schema print(sequence.output_schema.model_json_schema()) # Show inferred output schema print(sequence.invoke(2)) # invoke the sequence (note the retry above!!) ``` Debugging and tracing ===================== As the chains get longer, it can be useful to be able to see intermediate results to debug and trace the chain. You can set the global debug flag to True to enable debug output for all chains: ```python from langchain_core.globals import set_debug set_debug(True) ``` Alternatively, you can pass existing or custom callbacks to any given chain: ```python from langchain_core.tracers import ConsoleCallbackHandler chain.invoke(..., config={"callbacks": [ConsoleCallbackHandler()]}) ``` For a UI (and much more) checkout [LangSmith](https://docs.langchain.com/langsmith/home). str | NonenameNrhcU(aUnO[US5(aUR(a URnOcURn[US5(a:SUR;a*URSbURSRnO URnU(a@USR 5(aX1R 5-$US-UR5-$U$)zGet the name of the `Runnable`. Args: suffix: An optional suffix to append to the name. name: An optional name to use instead of the `Runnable`'s name. Returns: The name of the `Runnable`. rh__pydantic_generic_metadata__originr_)hasattrrh __class__rk__name__isuppertitlelower)selfsuffixrhname_clss W/home/james-whalen/.local/lib/python3.13/site-packages/langchain_core/runnables/base.pyget_nameRunnable.get_names E T6 " "tyyIIE..C3 A AA55h?K99(CLL  Qx!!||~--3;/ / cURR5HTn[US5(dMURnSU;dM*[ US5[ :XdMB[ SUSS5s $ URRH;n[U5nU(dM[ U5[ :XdM,[ SUS5s $ SUR5S3n[U5e)zInput type. The type of input this `Runnable` accepts specified as a type annotation. Raises: TypeError: If the input type cannot be inferred. rkargs type[Input]r Runnable z` doesn't have an inferable InputType. Override the InputType property to specify the input type. romrornrklen_RUNNABLE_GENERIC_NUM_ARGSr__orig_bases__rry TypeErrorrtbasemetadatarw type_argsmsgs rx InputTypeRunnable.InputType+sNN&&(Dt<====h&HV,-1KK x/?/BCC)>>00C  IyS^/IIM9Q<881  ()I I nr{cURR5HTn[US5(dMURnSU;dM*[ US5[ :XdMB[ SUSS5s $ URRH;n[U5nU(dM[ U5[ :XdM,[ SUS5s $ SUR5S3n[U5e)zOutput Type. The type of output this `Runnable` produces specified as a type annotation. Raises: TypeError: If the output type cannot be inferred. rkr} type[Output]rzc doesn't have an inferable OutputType. Override the OutputType property to specify the output type.rrs rx OutputTypeRunnable.OutputTypeNsNN&&(Dt<====h&HV,-1KK0@0CDD)>>00C  IyS^/IINIaL991  ()K K nr{c"UR5$)zHThe type of input this `Runnable` accepts specified as a Pydantic model.)get_input_schemarts rx input_schemaRunnable.input_schemams$$&&r{cURn[R"U5(a,[U[5(d[ U[ 5(aU$[URS5UURRS9$)aGet a Pydantic model that can be used to validate input to the `Runnable`. `Runnable` objects that leverage the `configurable_fields` and `configurable_alternatives` methods will have a dynamic input schema that depends on which configuration the `Runnable` is invoked with. This method allows to get an input schema for a specific configuration. Args: config: A config to use when generating the schema. Returns: A Pydantic model that can be used to validate input. r9root module_name) rinspectisclass isinstancer issubclassrrRryro __module__rtconfig root_types rxrRunnable.get_input_schemarsi$NN  OOI & &y,779i00  MM' "11  r{c@URU5R5$)aGet a JSON schema that represents the input to the `Runnable`. Args: config: A config to use when generating the schema. Returns: A JSON schema that represents the input to the `Runnable`. Example: ```python from langchain_core.runnables import RunnableLambda def add_one(x: int) -> int: return x + 1 runnable = RunnableLambda(add_one) print(runnable.get_input_jsonschema()) ``` !!! version-added "Added in `langchain-core` 0.3.0" )rmodel_json_schemartrs rxget_input_jsonschemaRunnable.get_input_jsonschemas8$$V,>>@@r{c"UR5$)z[Output schema. The type of output this `Runnable` produces specified as a Pydantic model. )get_output_schemars rx output_schemaRunnable.output_schemas %%''r{cURn[R"U5(a,[U[5(d[ U[ 5(aU$[URS5UURRS9$)aGet a Pydantic model that can be used to validate output to the `Runnable`. `Runnable` objects that leverage the `configurable_fields` and `configurable_alternatives` methods will have a dynamic output schema that depends on which configuration the `Runnable` is invoked with. This method allows to get an output schema for a specific configuration. Args: config: A config to use when generating the schema. Returns: A Pydantic model that can be used to validate output. r:r) rrrrrrrrRryrorrs rxrRunnable.get_output_schemasi$OO  OOI & &y,779i00  MM( #11  r{c@URU5R5$)aGet a JSON schema that represents the output of the `Runnable`. Args: config: A config to use when generating the schema. Returns: A JSON schema that represents the output of the `Runnable`. Example: ```python from langchain_core.runnables import RunnableLambda def add_one(x: int) -> int: return x + 1 runnable = RunnableLambda(add_one) print(runnable.get_output_jsonschema()) ``` !!! version-added "Added in `langchain-core` 0.3.0" )rrrs rxget_output_jsonschemaRunnable.get_output_jsonschemas8%%f-??AAr{c/$)z-List configurable fields for this `Runnable`.rs rx config_specsRunnable.config_specss  r{includecU=(d /nURnU(aY[SUVs0sHCnURUR[ UR UR URS94_ME snS9OSn0U(aSUS40O0E[[5R5VVVs0sH(upVXQVs/sH owS:wdM UPM sn;dM$XVS4_M* snnnEn[URS5US9$s snfs snfs snnnf)a>The type of config this `Runnable` accepts specified as a Pydantic model. To mark a field as configurable, see the `configurable_fields` and `configurable_alternatives` methods. Args: include: A list of fields to include in the config schema. Returns: A Pydantic model that can be used to validate config. Configurable)rr descriptionfield_definitionsN configurableConfig) rrRid annotationr defaultrhrrr)itemsry) rtrrspecr field_name field_typei all_fieldss rx config_schemaRunnable.config_schema s%-R((  !-#!- GG LL tGWGW !-#  $ 9Et 452 /=^.L.R.R.T.T*JW!LW^8K!W!LL/ ...T t}}X6*UU/#("Ms*A C9 + D 7 C>C> D D >D c<URUS9R5$)aGet a JSON schema that represents the config of the `Runnable`. Args: include: A list of fields to include in the config schema. Returns: A JSON schema that represents the config of the `Runnable`. !!! version-added "Added in `langchain-core` 0.3.0" r)rr)rtrs rxget_config_jsonschemaRunnable.get_config_jsonschema6s!!!'!2DDFFr{cSSKJn U"5nURURU55nURX(aURS5OSS9nURURU55nURXE5 URXV5 U$![a, UR[ UR S555nNf=f![a, UR[ UR S555nNf=f)z1Return a graph representation of this `Runnable`.rrYr9rN)rr:) langchain_core.runnables.graphrZadd_noderrrRrygetradd_edge)rtrrZgraph input_node runnable_node output_nodes rx get_graphRunnable.get_graphFs 9 Q(=(=f(EFJ V6::j1'  S..)?)?)GHK z1 }2  Q g8N(OPJ Q S..x9P)QRK Ss# B C3CC3D  D cSSKJn URUS9RR 5Vs/sH,n[ UR U5(dM UR PM. sn$s snf)z1Return a list of prompts used by this `Runnable`.rrUr)langchain_core.prompts.baserVrnodesvaluesrdata)rtrrVnodes rx get_promptsRunnable.get_prompts[s^ Cf5;;BBD D$))%78 DIID   s A&A&c,[U[U55$)zRunnable "or" operator. Compose this `Runnable` with another object to create a `RunnableSequence`. Args: other: Another `Runnable` or a `Runnable`-like object. Returns: A new `Runnable`. RunnableSequencecoerce_to_runnablertothers rx__or__Runnable.__or__hs& &8&?@@r{c,[[U5U5$)zRunnable "reverse-or" operator. Compose this `Runnable` with another object to create a `RunnableSequence`. Args: other: Another `Runnable` or a `Runnable`-like object. Returns: A new `Runnable`. rrs rx__ror__Runnable.__ror__}s& 25 94@@r{c [U/UQ7SU06$)aPipe `Runnable` objects. Compose this `Runnable` with `Runnable`-like objects to make a `RunnableSequence`. Equivalent to `RunnableSequence(self, *others)` or `self | others[0] | ...` Example: ```python from langchain_core.runnables import RunnableLambda def add_one(x: int) -> int: return x + 1 def mul_two(x: int) -> int: return x * 2 runnable_1 = RunnableLambda(add_one) runnable_2 = RunnableLambda(mul_two) sequence = runnable_1.pipe(runnable_2) # Or equivalently: # sequence = runnable_1 | runnable_2 # sequence = RunnableSequence(first=runnable_1, last=runnable_2) sequence.invoke(1) await sequence.ainvoke(1) # -> 4 sequence.batch([1, 2, 3]) await sequence.abatch([1, 2, 3]) # -> [4, 6, 8] ``` Args: *others: Other `Runnable` or `Runnable`-like objects to compose name: An optional name for the resulting `RunnableSequence`. Returns: A new `Runnable`. rh)r)rtrhotherss rxpipe Runnable.pipes^ 9v9D99r{c"SSKJn X"U5-$)aPick keys from the output `dict` of this `Runnable`. !!! example "Pick a single key" ```python import json from langchain_core.runnables import RunnableLambda, RunnableMap as_str = RunnableLambda(str) as_json = RunnableLambda(json.loads) chain = RunnableMap(str=as_str, json=as_json) chain.invoke("[1, 2, 3]") # -> {"str": "[1, 2, 3]", "json": [1, 2, 3]} json_only_chain = chain.pick("json") json_only_chain.invoke("[1, 2, 3]") # -> [1, 2, 3] ``` !!! example "Pick a list of keys" ```python from typing import Any import json from langchain_core.runnables import RunnableLambda, RunnableMap as_str = RunnableLambda(str) as_json = RunnableLambda(json.loads) def as_bytes(x: Any) -> bytes: return bytes(x, "utf-8") chain = RunnableMap( str=as_str, json=as_json, bytes=RunnableLambda(as_bytes) ) chain.invoke("[1, 2, 3]") # -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"} json_and_bytes_chain = chain.pick(["json", "bytes"]) json_and_bytes_chain.invoke("[1, 2, 3]") # -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"} ``` Args: keys: A key or list of keys to pick from the output dict. Returns: a new `Runnable`. r) RunnablePick)$langchain_core.runnables.passthroughr)rtkeysrs rxpick Runnable.picksv Fl4(((r{c ^SSKJn X"[[[[ 4"U55-$)aAssigns new fields to the `dict` output of this `Runnable`. ```python from langchain_core.language_models.fake import FakeStreamingListLLM from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import SystemMessagePromptTemplate from langchain_core.runnables import Runnable from operator import itemgetter prompt = ( SystemMessagePromptTemplate.from_template("You are a nice assistant.") + "{question}" ) model = FakeStreamingListLLM(responses=["foo-lish"]) chain: Runnable = prompt | model | {"str": StrOutputParser()} chain_with_assign = chain.assign(hello=itemgetter("str") | model) print(chain_with_assign.input_schema.model_json_schema()) # {'title': 'PromptInput', 'type': 'object', 'properties': {'question': {'title': 'Question', 'type': 'string'}}} print(chain_with_assign.output_schema.model_json_schema()) # {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties': {'str': {'title': 'Str', 'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}} ``` Args: **kwargs: A mapping of keys to `Runnable` or `Runnable`-like objects that will be invoked with the entire output dict of this `Runnable`. Returns: A new `Runnable`. r)RunnableAssign)rrRunnableParalleldictstrr)rtkwargsrs rxassignRunnable.assigns,V Hn%5d38n%Ef%MNNNr{c gaTransform a single input into an output. Args: input: The input to the `Runnable`. config: A config to use when invoking the `Runnable`. The config supports standard keys like `'tags'`, `'metadata'` for tracing purposes, `'max_concurrency'` for controlling how much work to do in parallel, and other keys. Please refer to `RunnableConfig` for more details. Returns: The output of the `Runnable`. Nrrtinputrrs rxinvokeRunnable.invoke3sr{cL# [X RX40UD6IShvN $N7fr)r3rrs rxainvokeRunnable.ainvokeJs#*%V[[%R6RRRRs $"$Freturn_exceptionsc F^^^U(d/$[U[U55nSUUU4Sjjn[U5S:Xa[SU"USUS5/5$[US5n[S[ UR XaU555sSSS5 $!,(df  g=f)aaDefault implementation runs invoke in parallel using a thread pool executor. The default implementation of batch works well for IO bound runnables. Subclasses must override this method if they can batch more efficiently; e.g., if the underlying `Runnable` uses an API which supports a batch mode. Args: inputs: A list of inputs to the `Runnable`. config: A config to use when invoking the `Runnable`. The config supports standard keys like `'tags'`, `'metadata'` for tracing purposes, `'max_concurrency'` for controlling how much work to do in parallel, and other keys. Please refer to `RunnableConfig` for more details. return_exceptions: Whether to return exceptions instead of raising them. **kwargs: Additional keyword arguments to pass to the `Runnable`. Returns: A list of outputs from the `Runnable`. c>T(aTR"X40TD6$TR"X40TD6$![a nUsSnA$SnAff=fNr Exception)input_rerrrts rxrRunnable.batch..invokesL ;;v@@@{{6T(aTR"X40TD6nX4$TR"X40TD6nX4$![anUnSnAX4$SnAff=fr r rrroutrrrrts rxr+Runnable.batch_as_completed..invokese!.2kk&.SF.SC 8Okk&;F;8O !C8O s6 AA  ArrFstrict return_whenrintrr9rr)rztuple[int, Output | Exception]) r/rr0 enumeratezipsubmitrrpopresultcancel) rtrrrrrrrrrfuturesdonefutures ` `` rxrrs=8 !&#f+6  ! +9  +   v;! F1Iwqz2 2  $WQZ 0H,5SQV5W+X+X'A'6:+X   $$(o$NMD"hhj//11$g &FMMO&1 0&FMMO&1 0sOA E %D;=!D D;#AD*D3D; E D;D88D;; E E c^^^# U(d/$[U[U55nSUUU4Sjjn[XaU5n[USR S5/UQ76IShvN $N7f)abDefault implementation runs `ainvoke` in parallel using `asyncio.gather`. The default implementation of `batch` works well for IO bound runnables. Subclasses must override this method if they can batch more efficiently; e.g., if the underlying `Runnable` uses an API which supports a batch mode. Args: inputs: A list of inputs to the `Runnable`. config: A config to use when invoking the `Runnable`. The config supports standard keys like `'tags'`, `'metadata'` for tracing purposes, `'max_concurrency'` for controlling how much work to do in parallel, and other keys. Please refer to `RunnableConfig` for more details. return_exceptions: Whether to return exceptions instead of raising them. **kwargs: Additional keyword arguments to pass to the `Runnable`. Returns: A list of outputs from the `Runnable`. c># T(aTR"X40TD6IShvN $TR"X40TD6IShvN $N![a nUsSnA$SnAff=fN7fr rr )valuerrrrrts rxr Runnable.abatch..ainvoke s_ !%e!Fv!FFF"\\%B6BBB G HCsMAAAAAAAA AAAAAArmax_concurrencyN)r9r9rr)rr)r/rrr?r)rtrrrrrrcoross` `` rxabatchRunnable.abatchs_>I!&#f+6 C CGW-,WQZ^^# T(aTR"X40TD6IShvN nX4$TR"X40TD6IShvN nX4$N%![anUnSnAX4$SnAff=fN$7fr r8r$s rx ainvoke_task2Runnable.abatch_as_completed..ainvoke_taskPs!48LL5*05/C8O!LLB6BB8O/!C8O CsIA'A AA A'A%A'A A"AA'A""A'Fr'r+) r/rrasyncio Semaphorer-r.r> as_completed) rtrrrrrr; semaphorerGrrr<coros ` `` rxr@rA,s8 !&#f+6?F'!*..):;D:IG%%o6t   ! +9  +  $(1VU1S'T  (U##F y,q&"A Ba0 1(U  ((/D* 0 sBC11C(9"C1C/C1c+6# UR"X40UD6v g7f)aUDefault implementation of `stream`, which calls `invoke`. Subclasses must override this method if they support streaming output. Args: input: The input to the `Runnable`. config: The config to use for the `Runnable`. **kwargs: Additional keyword arguments to pass to the `Runnable`. Yields: The output of the `Runnable`. N)rrs rxstreamRunnable.streamhs&kk%2622sc L# UR"X40UD6IShvN 7v gN7f)aWDefault implementation of `astream`, which calls `ainvoke`. Subclasses must override this method if they support streaming output. Args: input: The input to the `Runnable`. config: The config to use for the `Runnable`. **kwargs: Additional keyword arguments to pass to the `Runnable`. Yields: The output of the `Runnable`. N)rrs rxastreamRunnable.astream}s!&LL9&9999s $" $T)diffwith_streamed_output_list include_names include_types include_tags exclude_names exclude_types exclude_tagsc gr r rtrrrTrUrVrWrXrYrZr[rs rx astream_logRunnable.astream_logs&)r{)rUrVrWrXrYrZr[c gr rr]s rxr^r_s!$r{c v# [SUUUUU U SS9n [UUU4UU US.U D6ShvN n U 7v MN g7f)aStream all output from a `Runnable`, as reported to the callback system. This includes all inner runs of LLMs, Retrievers, Tools, etc. Output is streamed as Log objects, which include a list of Jsonpatch ops that describe how the state of the run has changed in each step, and the final state of the run. The Jsonpatch ops can be applied in order to construct state. Args: input: The input to the `Runnable`. config: The config to use for the `Runnable`. diff: Whether to yield diffs between each step or the current state. with_streamed_output_list: Whether to yield the `streamed_output` list. include_names: Only include logs with these names. include_types: Only include logs with these types. include_tags: Only include logs with these tags. exclude_names: Exclude logs with these names. exclude_types: Exclude logs with these types. exclude_tags: Exclude logs with these tags. **kwargs: Additional keyword arguments to pass to the `Runnable`. Yields: A `RunLogPatch` or `RunLog` object. Foriginal) auto_closerVrWrXrYrZr[_schema_format)rTrOrUNrJ)rtrrrTrUrVrWrXrYrZr[rrOitems rxr^r_soT*''%''%%  6    &?   $J  s#9757979v2)versionrVrWrXrYrZr[c B# US:Xa[UU4UUUUUUU S.U D6n O)US:Xa[UU4UUUUUUU S.U D6n O Sn [U 5e[U 5IShvN U ShvN n U 7v MNN SSS5IShvN g!,IShvN (df  g=f7f)a$Generate a stream of events. Use to create an iterator over `StreamEvent` that provide real-time information about the progress of the `Runnable`, including `StreamEvent` from intermediate results. A `StreamEvent` is a dictionary with the following schema: - `event`: Event names are of the format: `on_[runnable_type]_(start|stream|end)`. - `name`: The name of the `Runnable` that generated the event. - `run_id`: Randomly generated ID associated with the given execution of the `Runnable` that emitted the event. A child `Runnable` that gets invoked as part of the execution of a parent `Runnable` is assigned its own unique ID. - `parent_ids`: The IDs of the parent runnables that generated the event. The root `Runnable` will have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list. - `tags`: The tags of the `Runnable` that generated the event. - `metadata`: The metadata of the `Runnable` that generated the event. - `data`: The data associated with the event. The contents of this field depend on the type of event. See the table below for more details. Below is a table that illustrates some events that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table. !!! note This reference table is for the v2 version of the schema. | event | name | chunk | input | output | | ---------------------- | -------------------- | ----------------------------------- | ------------------------------------------------- | --------------------------------------------------- | | `on_chat_model_start` | `'[model name]'` | | `{"messages": [[SystemMessage, HumanMessage]]}` | | | `on_chat_model_stream` | `'[model name]'` | `AIMessageChunk(content="hello")` | | | | `on_chat_model_end` | `'[model name]'` | | `{"messages": [[SystemMessage, HumanMessage]]}` | `AIMessageChunk(content="hello world")` | | `on_llm_start` | `'[model name]'` | | `{'input': 'hello'}` | | | `on_llm_stream` | `'[model name]'` | `'Hello' ` | | | | `on_llm_end` | `'[model name]'` | | `'Hello human!'` | | | `on_chain_start` | `'format_docs'` | | | | | `on_chain_stream` | `'format_docs'` | `'hello world!, goodbye world!'` | | | | `on_chain_end` | `'format_docs'` | | `[Document(...)]` | `'hello world!, goodbye world!'` | | `on_tool_start` | `'some_tool'` | | `{"x": 1, "y": "2"}` | | | `on_tool_end` | `'some_tool'` | | | `{"x": 1, "y": "2"}` | | `on_retriever_start` | `'[retriever name]'` | | `{"query": "hello"}` | | | `on_retriever_end` | `'[retriever name]'` | | `{"query": "hello"}` | `[Document(...), ..]` | | `on_prompt_start` | `'[template_name]'` | | `{"question": "hello"}` | | | `on_prompt_end` | `'[template_name]'` | | `{"question": "hello"}` | `ChatPromptValue(messages: [SystemMessage, ...])` | In addition to the standard events, users can also dispatch custom events (see example below). Custom events will be only be surfaced with in the v2 version of the API! A custom event has following format: | Attribute | Type | Description | | ----------- | ------ | --------------------------------------------------------------------------------------------------------- | | `name` | `str` | A user defined name for the event. | | `data` | `Any` | The data associated with the event. This can be anything, though we suggest making it JSON serializable. | Here are declarations associated with the standard events shown above: `format_docs`: ```python def format_docs(docs: list[Document]) -> str: '''Format the docs.''' return ", ".join([doc.page_content for doc in docs]) format_docs = RunnableLambda(format_docs) ``` `some_tool`: ```python @tool def some_tool(x: int, y: str) -> dict: '''Some_tool.''' return {"x": x, "y": y} ``` `prompt`: ```python template = ChatPromptTemplate.from_messages( [ ("system", "You are Cat Agent 007"), ("human", "{question}"), ] ).with_config({"run_name": "my_template", "tags": ["my_template"]}) ``` !!! example ```python from langchain_core.runnables import RunnableLambda async def reverse(s: str) -> str: return s[::-1] chain = RunnableLambda(func=reverse) events = [ event async for event in chain.astream_events("hello", version="v2") ] # Will produce the following events # (run_id, and parent_ids has been omitted for brevity): [ { "data": {"input": "hello"}, "event": "on_chain_start", "metadata": {}, "name": "reverse", "tags": [], }, { "data": {"chunk": "olleh"}, "event": "on_chain_stream", "metadata": {}, "name": "reverse", "tags": [], }, { "data": {"output": "olleh"}, "event": "on_chain_end", "metadata": {}, "name": "reverse", "tags": [], }, ] ``` ```python title="Dispatch custom event" from langchain_core.callbacks.manager import ( adispatch_custom_event, ) from langchain_core.runnables import RunnableLambda, RunnableConfig import asyncio async def slow_thing(some_input: str, config: RunnableConfig) -> str: """Do something that takes a long time.""" await asyncio.sleep(1) # Placeholder for some slow operation await adispatch_custom_event( "progress_event", {"message": "Finished step 1 of 3"}, config=config # Must be included for python < 3.10 ) await asyncio.sleep(1) # Placeholder for some slow operation await adispatch_custom_event( "progress_event", {"message": "Finished step 2 of 3"}, config=config # Must be included for python < 3.10 ) await asyncio.sleep(1) # Placeholder for some slow operation return "Done" slow_thing = RunnableLambda(slow_thing) async for event in slow_thing.astream_events("some_input", version="v2"): print(event) ``` Args: input: The input to the `Runnable`. config: The config to use for the `Runnable`. version: The version of the schema to use, either `'v2'` or `'v1'`. Users should use `'v2'`. `'v1'` is for backwards compatibility and will be deprecated in `0.4.0`. No default will be assigned until the API is stabilized. custom events will only be surfaced in `'v2'`. include_names: Only include events from `Runnable` objects with matching names. include_types: Only include events from `Runnable` objects with matching types. include_tags: Only include events from `Runnable` objects with matching tags. exclude_names: Exclude events from `Runnable` objects with matching names. exclude_types: Exclude events from `Runnable` objects with matching types. exclude_tags: Exclude events from `Runnable` objects with matching tags. **kwargs: Additional keyword arguments to pass to the `Runnable`. These will be passed to `astream_log` as this implementation of `astream_events` is built on top of `astream_log`. Yields: An async stream of `StreamEvent`. Raises: NotImplementedError: If the version is not `'v1'` or `'v2'`. rf)rrVrWrXrYrZr[v1zAOnly versions "v1" and "v2" of the schema is currently supported.N)rIrHNotImplementedErrorrO)rtrrrgrVrWrXrYrZr[r event_streamrevents rxastream_eventsRunnable.astream_eventssd d?< ++)++)  L_= ++)++)  LVC%c* *L))+ e * |*)))))slABA.BBA2"A0#A2&B.B0A22B3 B>B?BB B BBc+# SnUHnU(dUnSnMWU-nM U(aUR"WU40UD6ShvN gg![a UnMNf=fN7f)aTransform inputs to outputs. Default implementation of transform, which buffers input and calls `astream`. Subclasses must override this method if they can start producing output while input is still being generated. Args: input: An iterator of inputs to the `Runnable`. config: The config to use for the `Runnable`. **kwargs: Additional keyword arguments to pass to the `Runnable`. Yields: The output of the `Runnable`. FTN)rrO)rtrrr got_first_valichunkfinals rx transformRunnable.transformsq. F! $ #!FNE" {{5&;F; ; ; !#"E# S[UTTTS9/0$N callbacks)rrrrrNrrrrs rx)Runnable.with_listeners..s$+#)%-#)%- " r{rzconfig_factoriesr{rtrrrs ```rxwith_listenersRunnable.with_listenerss p    r{c*^^^[UUUU4Sj/S9$)aM Bind async lifecycle listeners to a `Runnable`. Returns a new `Runnable`. The Run object contains information about the run, including its `id`, `type`, `input`, `output`, `error`, `start_time`, `end_time`, and any tags or metadata added to the run. Args: on_start: Called asynchronously before the `Runnable` starts running, with the `Run` object. on_end: Called asynchronously after the `Runnable` finishes running, with the `Run` object. on_error: Called asynchronously if the `Runnable` throws an error, with the `Run` object. Returns: A new `Runnable` with the listeners bound. Example: ```python from langchain_core.runnables import RunnableLambda, Runnable from datetime import datetime, timezone import time import asyncio def format_t(timestamp: float) -> str: return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat() async def test_runnable(time_to_sleep: int): print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}") await asyncio.sleep(time_to_sleep) print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}") async def fn_start(run_obj: Runnable): print(f"on start callback starts at {format_t(time.time())}") await asyncio.sleep(3) print(f"on start callback ends at {format_t(time.time())}") async def fn_end(run_obj: Runnable): print(f"on end callback starts at {format_t(time.time())}") await asyncio.sleep(2) print(f"on end callback ends at {format_t(time.time())}") runnable = RunnableLambda(test_runnable).with_alisteners( on_start=fn_start, on_end=fn_end ) async def concurrent_runs(): await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3)) asyncio.run(concurrent_runs()) # Result: # on start callback starts at 2025-03-01T07:05:22.875378+00:00 # on start callback starts at 2025-03-01T07:05:22.875495+00:00 # on start callback ends at 2025-03-01T07:05:25.878862+00:00 # on start callback ends at 2025-03-01T07:05:25.878947+00:00 # Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00 # Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00 # Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00 # on end callback starts at 2025-03-01T07:05:27.882360+00:00 # Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00 # on end callback starts at 2025-03-01T07:05:28.882428+00:00 # on end callback ends at 2025-03-01T07:05:29.883893+00:00 # on end callback ends at 2025-03-01T07:05:30.884831+00:00 ``` c">S[UTTTS9/0$r)rMrs rxr*Runnable.with_alisteners..s$0#)%-#)%- " r{rr{rs ```rxwith_alistenersRunnable.with_alistenerss b    r{) input_type output_typec[UUU0S9$)zBind input and output types to a `Runnable`, returning a new `Runnable`. Args: input_type: The input type to bind to the `Runnable`. output_type: The output type to bind to the `Runnable`. Returns: A new `Runnable` with the types bound. )rzcustom_input_typecustom_output_typerr{rtrrs rx with_typesRunnable.with_types,s(*   r{)retry_if_exception_typewait_exponential_jitterexponential_jitter_paramsstop_after_attemptc &SSKJn U"U00UUUUS9$)aQCreate a new `Runnable` that retries the original `Runnable` on exceptions. Args: retry_if_exception_type: A tuple of exception types to retry on. wait_exponential_jitter: Whether to add jitter to the wait time between retries. stop_after_attempt: The maximum number of attempts to make before giving up. exponential_jitter_params: Parameters for `tenacity.wait_exponential_jitter`. Namely: `initial`, `max`, `exp_base`, and `jitter` (all `float` values). Returns: A new `Runnable` that retries the original `Runnable` on exceptions. Example: ```python from langchain_core.runnables import RunnableLambda count = 0 def _lambda(x: int) -> None: global count count = count + 1 if x == 1: raise ValueError("x is 1") else: pass runnable = RunnableLambda(_lambda) try: runnable.with_retry( stop_after_attempt=2, retry_if_exception_type=(ValueError,), ).invoke(1) except ValueError: pass assert count == 2 ``` r) RunnableRetry)rzrrretry_exception_typesrmax_attempt_numberr)langchain_core.runnables.retryr)rtrrrrrs rx with_retryRunnable.with_retryBs+h A"9$;1&?  r{c[US9$)aReturn a new `Runnable` that maps a list of inputs to a list of outputs. Calls `invoke` with each input. Returns: A new `Runnable` that maps a list of inputs to a list of outputs. Example: ```python from langchain_core.runnables import RunnableLambda def _lambda(x: int) -> int: return x + 1 runnable = RunnableLambda(_lambda) print(runnable.map().invoke([1, 2, 3])) # [2, 3, 4] ``` rz) RunnableEachrs rxr Runnable.maps*$''r{)exceptions_to_handle exception_keyc SSKJn U"UUUUS9$)a>Add fallbacks to a `Runnable`, returning a new `Runnable`. The new `Runnable` will try the original `Runnable`, and then each fallback in order, upon failures. Args: fallbacks: A sequence of runnables to try if the original `Runnable` fails. exceptions_to_handle: A tuple of exception types to handle. exception_key: If `string` is specified then handled exceptions will be passed to fallbacks as part of the input under the specified key. If `None`, exceptions will not be passed to fallbacks. If used, the base `Runnable` and its fallbacks must accept a dictionary as input. Returns: A new `Runnable` that will try the original `Runnable`, and then each Fallback in order, upon failures. Example: ```python from typing import Iterator from langchain_core.runnables import RunnableGenerator def _generate_immediate_error(input: Iterator) -> Iterator[str]: raise ValueError() yield "" def _generate(input: Iterator) -> Iterator[str]: yield from "foo bar" runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks( [RunnableGenerator(_generate)] ) print("".join(runnable.stream({}))) # foo bar ``` Args: fallbacks: A sequence of runnables to try if the original `Runnable` fails. exceptions_to_handle: A tuple of exception types to handle. exception_key: If `string` is specified then handled exceptions will be passed to fallbacks as part of the input under the specified key. If `None`, exceptions will not be passed to fallbacks. If used, the base `Runnable` and its fallbacks must accept a dictionary as input. Returns: A new `Runnable` that will try the original `Runnable`, and then each Fallback in order, upon failures. rrW)runnable fallbacksrr)"langchain_core.runnables.fallbacksrX)rtrrrrXs rxwith_fallbacksRunnable.with_fallbackss#F %!5'   r{c [U5n[U5nURUUUURS5=(d UR 5UR SS5S9n[ X8R5S9n [U 5n [SU R"[UUUU40UD65n SSS5 URW 5 U $!,(df  N!=f![an URU 5 eSn A ff=f)zCall with config. Helper method to transform an `Input` value to an `Output` value, with callbacks. Use this method to implement `invoke` in subclasses. run_namerun_idNrun_typerhrrr:)r,r.on_chain_startrryr0r2 get_childr4rrunr+ on_chain_end BaseExceptionon_chain_error) rtfuncrrr serializedrcallback_manager run_manager child_configcontextrvrs rx_call_with_configRunnable._call_with_configs&v&:6B&55  J':4==?::h- 6  ':O:O:QRL#L1WKK4#  !  2  $ $V ,M#21   & &q )  s0!"C&C)C CC C6C11C6c .# [U5n[U5nURUUUURS5=(d UR 5UR SS5S9IShvN n[ X8R5S9n [U 5n [XX840UD6n [X5IShvN n SSS5 URW 5IShvN U $NkN)!,(df  N-=fN![a n URU 5IShvN eSn A ff=f7f)zAsync call with config. Helper method to transform an `Input` value to an `Output` value, with callbacks. Use this method to implement `ainvoke` in subclasses. rrNrr)r,r-rrryr0r2rr4r*r=rrr)rtrrrrrrrrrrrMrvrs rx_acall_with_configRunnable._acall_with_configs (v&@H,;;  J':4==?::h- <   ':O:O:QRL#L1W4&9?(9'G!G 2**62 2 2M' "H 21 3  ,,Q/ / /  s~A#D%C&D+"C( C)C*C.C(6D C& DC C#C(&D( D2D D D  DD)rrc U(d/$[U[U55nUVs/sHn[U5PM n n[XUSS9V V Vs/sHOupnU R SU UUR S5=(d UR 5URSS5S9PMQ n n n n[U5(a7[X|SS9VV s/sHup[XR5S9PM sn nUS'[U5(aXS 'U"U40UD6nSn[XSS9HJunn[U[5(aU=(d UnURU5 M9URU5 ML U(dUc [!S U5$Ues snfs snn n fs sn nf!["aOnU HnURU5 M U(a([!S UVs/sHnUPM Os snfsn5sSnA$eSnAff=f) zTransform a list of inputs to a list of outputs, with callbacks. Helper method to transform an `Input` value to an `Output` value, with callbacks. Use this method to implement `invoke` in subclasses. Fr'Nrrrrrrr)r/rr.r.rrryr0r;r2rr<rr rrrr)rtrrrrrrrccallback_managersrr run_managersrmrvfirst_exceptionrr%rrms rx_batch_with_configRunnable._batch_with_configEs.I!&#f+6IPQAt}}zz(D1 , 5    "d##"%W5!I$!I!lln=!I$x #4(((4}%&+F+F15O$' U$K Sc9--&5&<O..s3,,S1 %L !O$;NF33! !MR  $ + **1- , N,?1Q,?@@   sNE: AE?$F #F$$F F G%+G  G G G%G  G%c ^^^# U(d/$[U[U55nUVs/sHn[U5PM n n[R"UU4Sj[ XUSS956IShvN n [ U5(a7[ XzSS9VV s/sHup[XR5S9PM sn nUS'[U5(aXS'U"U40UD6IShvN n Sn /n[ XSS9Hhunn[U[5(a-U =(d Un URURU55 MHURURU55 Mj [R"U6IShvN U(dU c [SU 5$U es snfGN$s sn nfNN,![ a[m[R"U4S jU 56IShvN U(a([SUVs/sHnTPM Os snfsn5sSm@$eSm@ff=f7f) zTransform a list of inputs to a list of outputs, with callbacks. Helper method to transform an `Input` value to an `Output` value, with callbacks. Use this method to implement `invoke` in subclasses. c 3># UHQupnURSUTURS5=(d TR5URSS5S9v MS g7f)Nrrrrrryr0).0rrrrrts rx /Runnable._abatch_with_config..s` 94$f!//%J/B4==?!::h5 09sAAFr'Nrrrrc3D># UHoRT5v M g7fr r)rrrs rxrrsP? %L ..%( ( ( O$;NF33! !WX ? $ 2" )! ..P<P  !N,?1Q,?@@   s#HF,H)F *H/F #F/'FFFBH+F,"HFH G?$ G:GG: G( ' G:3G?4H9G::G??Hc +2# URSS5n[US5upx[US5n Sn Sn Sn [U5n[ U5n U R SSS0UUR S5=(d UR5URS S5US 9n[X>R5S 9n[U5(aXS '[U5(aXS '[U5nUR"X(40UD6n[SUR5S5=n(aURUR U5nUR[U5nUv U (a U cUn O U U-n OUn M2!["a Un Sn Nf=f![$[&4a Of=fUH0nU (a$U cUn MU U-n M!["a Un Sn M*f=fUn M2 SSS5 O!,(df  O=fUR)XS9 g![*anUR-UU S9 eSnAff=f7f)zTransform a stream with config. Helper method to transform an `Iterator` of `Input` values into an `Iterator` of `Output` values, with callbacks. Use this to implement `stream` or `transform` in `Runnable` subclasses. defers_inputsFrcNTrrrrrhrrrrrc3h# UH(n[U[5(dM[SU5v M* g7frGNrrGrrhs rxr9Runnable._transform_stream_with_config.. /!5A%a)BC=8!< .==#**H H(/ D((C# 1+3/4 !C3?%3GL ,1L(1!C38L=B$:!C &}5/F,&.*0K>.9F.B #,>.4 8= 5>'- 0A211`  $ $\ $ F    & &q & =  sBHA G4"AG7(E> E*%E>*E;8E>:E;;E>>FGFG(F/-G/G=GG G G4 G!G4$H4 H>HHHc d# URSS5n[US5upx[US5IShvN n Sn Sn Sn [U5n[ U5n U R SSS0UUR S5=(d UR5URS S5US 9IShvN n[X>R5S 9n[U5(aXS '[U5(aXS '[U5nUR"X(40UD6n[SUR5S5=n(aUR!UR"U5nOUn[%[U5U5IShvN nU7v U (a U cUn O U U-n OUn M:GNSNN&![&a Un Sn Nf=f![(a Of=fUShvN nU (a$U cUn MU U-n M ![&a Un Sn M1f=fUn M9 SSS5 O!,(df  O=fUR+XS9IShvN O,![,anUR/UU S9IShvN eSnAff=fWb,[1US5(aUR35IShvN ggg!Wb,[1US5(aUR35IShvN fff=f7f)zTransform a stream with config. Helper method to transform an Async `Iterator` of `Input` values into an Async `Iterator` of `Output` values, with callbacks. Use this to implement `astream` or `atransform` in `Runnable` subclasses. rFrcNTrrrrrrrrc3h# UH(n[U[5(dM[SU5v M* g7frrrs rxr:Runnable._atransform_stream_with_config..l rrraclose)r0rPanextr,r-rrryr2rr;r<r4rrrtap_output_aiterrr=rStopAsyncIterationrrrrnr)rtrrrrrrrrrrrrrrrr iterator_rrrrqrs rx_atransform_stream_with_config'Runnable._atransform_stream_with_config5 s2 ?E: 26fa.*/0A4*H$H $&* !%v&@H,;;  bMJ':4==?::h-' <  < )':O:O:QRLk**#/x ";//(3}%#L1W#KK SFS %)!,!5!5 &> .>>#**I H )H&7h&Q Q# 1+3/4 !C3?%3GL ,1LQ%I D!R(1!C38L=B$:!C *$5 - -&,&.*0K>.9F.B #,>.4 8= 5>'- %6G211f**<*L L L  ,,Q{,C C C   M$H)E)E&&(((*F$y$H)E)E&&(((*F$s80J0FA/J0"F#J0(A H2AG5 F #F $F ;F F J0J0 F  FF FF  F-*G5,F--G52G+6F97G+;G5 GG5G$ G5#G$$G5, H5 H?HI<HI< I)I<H?=III< 'J03I64J0<(J-$J'%J--J0z1This API is in beta and may change in the future.message)rhr arg_typesc"SSKJn U"UUUUUS9$)ae Create a `BaseTool` from a `Runnable`. `as_tool` will instantiate a `BaseTool` with a name, description, and `args_schema` from a `Runnable`. Where possible, schemas are inferred from `runnable.get_input_schema`. Alternatively (e.g., if the `Runnable` takes a dict as input and the specific `dict` keys are not typed), the schema can be specified directly with `args_schema`. You can also pass `arg_types` to just specify the required arguments and their types. Args: args_schema: The schema for the tool. name: The name of the tool. description: The description of the tool. arg_types: A dictionary of argument names to types. Returns: A `BaseTool` instance. !!! example "`TypedDict` input" ```python from typing_extensions import TypedDict from langchain_core.runnables import RunnableLambda class Args(TypedDict): a: int b: list[int] def f(x: Args) -> str: return str(x["a"] * max(x["b"])) runnable = RunnableLambda(f) as_tool = runnable.as_tool() as_tool.invoke({"a": 3, "b": [1, 2]}) ``` !!! example "`dict` input, specifying schema via `args_schema`" ```python from typing import Any from pydantic import BaseModel, Field from langchain_core.runnables import RunnableLambda def f(x: dict[str, Any]) -> str: return str(x["a"] * max(x["b"])) class FSchema(BaseModel): """Apply a function to an integer and list of integers.""" a: int = Field(..., description="Integer") b: list[int] = Field(..., description="List of ints") runnable = RunnableLambda(f) as_tool = runnable.as_tool(FSchema) as_tool.invoke({"a": 3, "b": [1, 2]}) ``` !!! example "`dict` input, specifying schema via `arg_types`" ```python from typing import Any from langchain_core.runnables import RunnableLambda def f(x: dict[str, Any]) -> str: return str(x["a"] * max(x["b"])) runnable = RunnableLambda(f) as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]}) as_tool.invoke({"a": 3, "b": [1, 2]}) ``` !!! example "`str` input" ```python from langchain_core.runnables import RunnableLambda def f(x: str) -> str: return x + "a" def g(x: str) -> str: return x + "z" runnable = RunnableLambda(f) | g as_tool = runnable.as_tool() as_tool.invoke("b") ``` r)convert_runnable_to_tool) args_schemarhrr)langchain_core.toolsr)rtrrhrrrs rxas_toolRunnable.as_tool s%Z B' ##   r{rr rurgrhrgrrrr~rr)rtype[BaseModel]rRunnableConfig | Nonerr)rrrdict[str, Any]rzlist[ConfigurableFieldSpec])rSequence[str] | Nonerr)rrrrrrrrZ)rrrzlist[BasePromptTemplate]rzRunnable[Any, Other] | Callable[[Iterator[Any]], Iterator[Other]] | Callable[[AsyncIterator[Any]], AsyncIterator[Other]] | Callable[[Any], Other] | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any]r"RunnableSerializable[Input, Other]rzRunnable[Other, Any] | Callable[[Iterator[Other]], Iterator[Any]] | Callable[[AsyncIterator[Other]], AsyncIterator[Any]] | Callable[[Other], Any] | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any]rz#RunnableSerializable[Other, Output])rz-Runnable[Any, Other] | Callable[[Any], Other]rhrgrr")rzstr | list[str]rRunnableSerializable[Any, Any])rzRunnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]]rr$rr9rrrrrr: r list[Input]r,RunnableConfig | list[RunnableConfig] | Nonerboolr Any | Nonerr rSequence[Input]r0RunnableConfig | Sequence[RunnableConfig] | NonerLiteral[False]rrrzIterator[tuple[int, Output]] rr,rr-r Literal[True]rrr(Iterator[tuple[int, Output | Exception]] rr,rr-rr)rr*rr1 rr,rr-rr.rr*rz!AsyncIterator[tuple[int, Output]] rr,rr-rr0rr*r-AsyncIterator[tuple[int, Output | Exception]] rr,rr-rr)rr*rr5rr9rrrr*rIterator[Output]rr9rrrr*rAsyncIterator[Output])rrrrrTr0rUr)rVrrWrrXrrYrrZrr[rrrrzAsyncIterator[RunLogPatch])rrrrrTr.rUr)rVrrWrrXrrYrrZrr[rrrrzAsyncIterator[RunLog])rrrrrTr)rUr)rVrrWrrXrrYrrZrr[rrrrz2AsyncIterator[RunLogPatch] | AsyncIterator[RunLog])rrrrrgzLiteral['v1', 'v2']rVrrWrrXrrYrrZrr[rrrrAsyncIterator[StreamEvent]rIterator[Input]rrrr*rr8rAsyncIterator[Input]rrrr*rr:rrrRunnable[Input, Output]rrrrrrArDCallable[[Run], None] | Callable[[Run, RunnableConfig], None] | NonerrDrrDrrA)rAsyncListener | NonerrErrErrA)rztype[Input] | Nonerztype[Output] | NonerrA) rtuple[type[BaseException], ...]rr)rzExponentialJitterParams | Nonerr,rrA)rz#Runnable[list[Input], list[Output]])rz!Sequence[Runnable[Input, Output]]rrFrrgrz%RunnableWithFallbacksT[Input, Output]NN)rzCallable[[Input], Output] | Callable[[Input, CallbackManagerForChainRun], Output] | Callable[[Input, CallbackManagerForChainRun, RunnableConfig], Output]rr9rrrrgrdict[str, Any] | Nonerr*rr:)rzCallable[[Input], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun, RunnableConfig], Awaitable[Output]]rr9rrrrgrrHrr*rr:)rzCallable[[list[Input]], list[Exception | Output]] | Callable[[list[Input], list[CallbackManagerForChainRun]], list[Exception | Output]] | Callable[[list[Input], list[CallbackManagerForChainRun], list[RunnableConfig]], list[Exception | Output]]rr'rr(rr)rrgrr*rr)raCallable[[list[Input]], Awaitable[list[Exception | Output]]] | Callable[[list[Input], list[AsyncCallbackManagerForChainRun]], Awaitable[list[Exception | Output]]] | Callable[[list[Input], list[AsyncCallbackManagerForChainRun], list[RunnableConfig]], Awaitable[list[Exception | Output]]]rr'rr(rr)rrgrr*rr) rr=rzCallable[[Iterator[Input]], Iterator[Output]] | Callable[[Iterator[Input], CallbackManagerForChainRun], Iterator[Output]] | Callable[[Iterator[Input], CallbackManagerForChainRun, RunnableConfig], Iterator[Output]]rrrrgrr*rr8) rr?rzCallable[[AsyncIterator[Input]], AsyncIterator[Output]] | Callable[[AsyncIterator[Input], AsyncCallbackManagerForChainRun], AsyncIterator[Output]] | Callable[[AsyncIterator[Input], AsyncCallbackManagerForChainRun, RunnableConfig], AsyncIterator[Output]]rrrrgrr*rr:) rztype[BaseModel] | Nonerhrgrrgrzdict[str, type] | Nonerr]);rpr __qualname____firstlineno____doc____annotations__rypropertyrrrrrrrrrrrrrrrrrrrrrrrrr=r@rOrRr^rmrsrwr~rrrrr rrrrrrrrr r#betar__static_attributes__rr{rxrere|s CJ E$$$L  D<'' )-& %&  & R/3A+A A<(()-& %&  & R/3B+B B<@D)VX26G.G G ,/3  +  !  ALA ,A*ALA -A0 /:>/:/: , /:b=)~-OX-O ( -O^)-  &      2)-SS&S S  S4@D0U #( 0U0U=0U  0U  0U 0UdDH+ -2 ++A+ * +  + &++DH77A7 ) 7  7 277DH?$ #( ?$?$A?$  ?$  ?$ 2?$H@D.X #( .X.X=.X  .X  .X .X`DH0 -2 00A0 * 0  0 +00DH<<A< ) <  < 7<<DH: #( ::A:  :  : 7:~)-33&3 3  30)-::&: :  :*)-) #*..2.2-1.2.2-1))&)  ) $( ),),)+),),)+)) $)) )-$ +/.2.2-1.2.2-1$$&$  $ $( $,$,$+$,$,$+$$ $$&)-A *..2.2-1.2.2-1AA&A  A $( A,A,A+A,A,A+AA <AL)-t (,.2.2-1.2.2-1tt&t % t , t,t+t,t,t+tt $tr)-+<+<&+< +<  +<`)-,#,&, ,  ,\ EH)- %   !  <F F  F F  !F V*.'+)- _ '_ % _ ' _ ! _ H*.+/  ' )  !  2EN<(,DH"# > "A> "& > $B >  >  !> @(6BK $( L 4L > L " L / L \2 $,0/P/  / & //*// /t $,0) ))&))*)) )l@DA"#(#A" A"A"=A" A"A"A" !A"^@DI"#(#I" I"I"=I" I"I" !I"" #I"j $`G`G `G&`G`G`G `G^ $i)$i) i)&i)i)i) i)V!TU/3t  "&,0 t +t  t  t * t  t Vt r{rec^\rSrSr%SrSrS\S'\"SS9r\ SU4Sjj5r SS jr S S S .SS jjr Sr U=r$)RunnableSerializablei z(Runnable that can be serialized to JSON.Nrgrhr)protected_namespacesc>[TU]5n[R"[5 UR 5US'SSS5 U$!,(df  U$=f)zgSerialize the `Runnable` to JSON. Returns: A JSON-serializable representation of the `Runnable`. rhN)superto_json contextlibsuppressr ry)rtdumpedros rxrURunnableSerializable.to_json# sF"   +!]]_F6N, , + s A Ac SSKJn [U5RnUH,nXC;dM SUSUSUR 53n[ U5e U"XS9$)aConfigure particular `Runnable` fields at runtime. Args: **kwargs: A dictionary of `ConfigurableField` instances to configure. Raises: ValueError: If a configuration key is not found in the `Runnable`. Returns: A new `Runnable` with the fields configured. !!! example ```python from langchain_core.runnables import ConfigurableField from langchain_openai import ChatOpenAI model = ChatOpenAI(max_tokens=20).configurable_fields( max_tokens=ConfigurableField( id="output_token_number", name="Max tokens in the output", description="The maximum number of tokens in the output", ) ) # max_tokens = 20 print( "max_tokens_20: ", model.invoke("tell me something about chess").content ) # max_tokens = 200 print( "max_tokens_200: ", model.with_config(configurable={"output_token_number": 200}) .invoke("tell me something about chess") .content, ) ``` r)RunnableConfigurableFieldszConfiguration key z not found in z: available keys are )rfields)%langchain_core.runnables.configurabler[type model_fieldsr ValueError)rtrr[r_keyrs rxconfigurable_fields(RunnableSerializable.configurable_fields0 soV Dz.. C&(^D6B**6*;*;*=)>@!o% *$FFr{rF) default_key prefix_keysc "SSKJn U"UUUUUS9$)aConfigure alternatives for `Runnable` objects that can be set at runtime. Args: which: The `ConfigurableField` instance that will be used to select the alternative. default_key: The default key to use if no alternative is selected. prefix_keys: Whether to prefix the keys with the `ConfigurableField` id. **kwargs: A dictionary of keys to `Runnable` instances or callables that return `Runnable` instances. Returns: A new `Runnable` with the alternatives configured. !!! example ```python from langchain_anthropic import ChatAnthropic from langchain_core.runnables.utils import ConfigurableField from langchain_openai import ChatOpenAI model = ChatAnthropic( model_name="claude-sonnet-4-5-20250929" ).configurable_alternatives( ConfigurableField(id="llm"), default_key="anthropic", openai=ChatOpenAI(), ) # uses the default model ChatAnthropic print(model.invoke("which organization created you?").content) # uses ChatOpenAI print( model.with_config(configurable={"llm": "openai"}) .invoke("which organization created you?") .content ) ``` r) RunnableConfigurableAlternatives)whichr alternativesrdre)r]rg)rtrhrdrerrgs rxconfigurable_alternatives.RunnableSerializable.configurable_alternativesj s&` 0##   r{)rz0SerializedConstructor | SerializedNotImplemented)rr6r#RunnableSerializable[Input, Output]) rhr7rdrrer)rz?Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]rrl)rprrIrJrKrhrLr model_configr"rUrbrjrO __classcell__ros@rxrQrQ s2D* L   8G,8G ,8G|%! :  :  :  : R :  -: : r{rQc SSKJnJn USn[U5S:XaUR U5$[ XB5(a[ USSU5n[U[5(dh[SURR5VVs0sH8upgXdRR;dM XgRUR4_M: snnS9$O[ XC5(a[ USSU5$UR U5$s snnf)NrrrrRunnableSequenceInputr)rrrrrr_seq_input_schemarr!rRr_rmappersteps__rr)stepsrrrfirstnext_input_schemakvs rxrsrs s !HE 5zQ%%f--%((-eABi@+Y77"'!2 > > D D F# F 4 441A aii00 F# 8 E ( ( qrF33  ! !& ))#s C; (C; c >SSKJnJn USn[U5S:XaUR U5$[ XB5(aUR RU5n[USSU5n[U[5(d[S0URR5VVs0sHupxXxRUR4_M snnEURR5VVs0sHupxXxRUR4_M snnES9$O[ XC5(a[USSU5n[U[5(d[ UR ["5(a^[SURR5VVs0sH.upxXtR ;dMXxRUR4_M0 snnS9$URUR n [SU RU R4S9$URU5$s snnfs snnfs snnf)NrrqrRunnableSequenceOutputr)r)rrrrrrrtr_seq_output_schemarr!rRr_rrrrr) rvrrrlastmapper_output_schemaprev_output_schemaryrzfields rxr~r~ s 9D 5zQ$$V,,$''#{{<5LL!))44$K''33DII>E"(0@0@%--/P   ! !& ))9's!$H &$HH !H cl^\rSrSr%SrS\S'\"\S9rS\S'S\S 'S S S S S .S(U4S jjjr \ \ S)S j55r \ S*Sj5r\ \ S+Sj55r\"SS9r\ \ S,Sj55r\ \ S-Sj55r\ S.S/Sjj5r\ S.S/Sjj5r\ \ S0Sj55r\ S.S1Sjj5r\ S2Sj5r\ S3Sj5r\ S4Sj5r\ S.S5Sjj5r\ S.S6Sjj5r\ S.SS.S7Sjjj5r\ S.SS.S7S jjj5rS8S!jrS9S"jr \ S.S:S#jj5r!\ S.S;S$jj5r"\ S.Sri a4 Sequence of `Runnable` objects, where the output of one is the input of the next. **`RunnableSequence`** is the most important composition operator in LangChain as it is used in virtually every chain. A `RunnableSequence` can be instantiated directly or more commonly by using the `|` operator where either the left or right operands (or both) must be a `Runnable`. Any `RunnableSequence` automatically supports sync, async, batch. The default implementations of `batch` and `abatch` utilize threadpools and asyncio gather and will be faster than naive invocation of `invoke` or `ainvoke` for IO bound `Runnable`s. Batching is implemented by invoking the batch method on each component of the `RunnableSequence` in order. A `RunnableSequence` preserves the streaming properties of its components, so if all components of the sequence implement a `transform` method -- which is the method that implements the logic to map a streaming input to a streaming output -- then the sequence will be able to stream input to output! If any component of the sequence does not implement transform then the streaming will only begin after this component is run. If there are multiple blocking components, streaming begins after the last one. !!! note `RunnableLambdas` do not support `transform` by default! So if you need to use a `RunnableLambdas` be careful about where you place them in a `RunnableSequence` (if you need to use the `stream`/`astream` methods). If you need arbitrary logic and need streaming, you can subclass Runnable, and implement `transform` for whatever logic you need. Here is a simple example that uses simple functions to illustrate the use of `RunnableSequence`: ```python from langchain_core.runnables import RunnableLambda def add_one(x: int) -> int: return x + 1 def mul_two(x: int) -> int: return x * 2 runnable_1 = RunnableLambda(add_one) runnable_2 = RunnableLambda(mul_two) sequence = runnable_1 | runnable_2 # Or equivalently: # sequence = RunnableSequence(first=runnable_1, last=runnable_2) sequence.invoke(1) await sequence.ainvoke(1) sequence.batch([1, 2, 3]) await sequence.abatch([1, 2, 3]) ``` Here's an example that uses streams JSON output generated by an LLM: ```python from langchain_core.output_parsers.json import SimpleJsonOutputParser from langchain_openai import ChatOpenAI prompt = PromptTemplate.from_template( "In JSON format, give me a list of {topic} and their " "corresponding names in French, Spanish and in a " "Cat Language." ) model = ChatOpenAI() chain = prompt | model | SimpleJsonOutputParser() async for chunk in chain.astream({"topic": "colors"}): print("-") # noqa: T201 print(chunk, sep="", flush=True) # noqa: T201 ``` Runnable[Input, Any]rwdefault_factorylist[Runnable[Any, Any]]middlezRunnable[Any, Output]rN)rhrwrrc>/nU(dUbUbU/U=(d /-U/-nUHOn[U[5(aURUR5 M5UR [ U55 MQ [ U5[:a S[S[ U53n[U5e[T U])US[USS5USUS9 g)aZCreate a new `RunnableSequence`. Args: steps: The steps to include in the sequence. name: The name of the `Runnable`. first: The first `Runnable` in the sequence. middle: The middle `Runnable` objects in the sequence. last: The last `Runnable` in the sequence. Raises: ValueError: If the sequence has less than 2 steps. Nz$RunnableSequence must have at least z steps, got rrr|)rwrrrh) rrextendrvrrr_RUNNABLE_SEQUENCE_MIN_STEPSr`rT__init__r) rtrhrwrrrv steps_flatsteprros rxrRunnableSequence.__init__Y s(&( *t/?FLb1TF:JD$ 011!!$**-!!"4T":;  z?9 967S6TU!*o.0 S/ ! Q- 1R()B  r{c /SQ$z_Get the namespace of the LangChain object. Returns: `["langchain", "schema", "runnable"]` ) langchainschemarrrws rxget_lc_namespace!RunnableSequence.get_lc_namespace 32r{cLUR/URQURP$)z]All the `Runnable`s that make up the sequence in order. Returns: A list of `Runnable`s. )rwrrrs rxrvRunnableSequence.steps s# 4T[[4$))44r{cgz,Return `True` as this class is serializable.Trrs rxis_lc_serializable#RunnableSequence.is_lc_serializable r{Tarbitrary_types_allowedc.URR$z(The type of the input to the `Runnable`.)rwrrs rxrRunnableSequence.InputType szz###r{c.URR$)z)The type of the output of the `Runnable`.)rrrs rxrRunnableSequence.OutputType syy###r{c.[URU5$)Get the input schema of the `Runnable`. Args: config: The config to use. Returns: The input schema of the `Runnable`. )rsrvrs rxr!RunnableSequence.get_input_schema s!V44r{c.[URU5$)Get the output schema of the `Runnable`. Args: config: The config to use. Returns: The output schema of the `Runnable`. )r~rvrs rxr"RunnableSequence.get_output_schema s"$**f55r{c[URVVs/sHoRHo"PM M snn5$s snnf)[Get the config specs of the `Runnable`. Returns: The config specs of the `Runnable`. )rCrvr)rtrrs rxrRunnableSequence.config_specs s7'"jj Gjd5F5FTT5FTj G  Gs; cSSKJn U"5nURHnUR5nUR U5nX@R LaUR 5 X@RLaUR5 URU5upxU(dSUS3n [U 5eU(dMURXW5 M U$)Get the graph representation of the `Runnable`. Args: config: The config to use. Returns: The graph representation of the `Runnable`. Raises: ValueError: If a `Runnable` has no first or last node. rrYr has no first node) rrZrv last_noderrwtrim_first_nodertrim_last_noderr`r) rtrrZrrcurrent_last_node step_graphstep_first_nodermrs rxrRunnableSequence.get_graph s 9JJD % 1 /J::%**,99$))+!&j!9 O"!$'9: o%  0B r{cXSRS[UR555$)Nz | c3r# UH-upUS:Xa [U5O[[U5S5v M/ g7f)rz| N)reprrD)rrss rxr,RunnableSequence.__repr__.. s4 -AvDG#;DGT#J J-s57)joinr-rvrs rx__repr__RunnableSequence.__repr__ s){{ !$**-   r{c[U[5(ap[UR/URQURPURPURQURP7SUR =(d UR 06$[UR/URQURP[ U5P7SUR 06$Nrhrrrwrrrhrrs rxrRunnableSequence.__or__ s e- . .#       YY,%**  JJ [[  II  u %     r{c[U[5(ap[UR/URQURPURPURQURP7SUR =(d UR 06$[[ U5UR/URQURP7SUR 06$rrrs rxrRunnableSequence.__ror__ s e- . .#       ZZ,499  u % JJ [[  II     r{c [U5n[U5nURSUURS5=(d UR 5UR SS5S9nUn[ UR5Hxupx[X%RSUS-35S9n[U5n US:XaU R"URXb40UD6nOU RURXb5nSSS5 Mz URU5 [SU5$!,(df  M=f![an UR!U 5 eSn A ff=f) Nrrrhr seq:step:rrrr:)r,r.rrryr0r-rvr2rr4rrrrrr) rtrrrrrrrrrrs rxrRunnableSequence.invoke4 s- v&:6B&55  J':4==?::h- 6  *$TZZ0%&;&;iAw>Ec # [U5n[U5nURSUURS5=(d UR 5UR SS5S9IShvN nUn[ UR5Hupx[X%RSUS-35S9n[U5n US:Xa#[R"URXb40UD6n O![R"URXb5n [U "5U SS 9IShvN nSSS5 M URU5IShvN [!S U5$NN7!,(df  M=fN&!["a n UR%U 5IShvN eSn A ff=f7f) NrrrrrrrT create_taskr:)r,r-rrryr0r-rvr2rr4 functoolspartialrr=rrrr) rtrrrrrrrrrpartrs rxrRunnableSequence.ainvokeX sjv&@H,;;  J':4==?::h- <   *$TZZ0%&;&;iAwt}}zz(D1   '   V " ;=!%.tzz%:MG$)W#6&#6aCT:T#6#&"ZZ+.nU*S*S 9 *S4= #L% H4  4 O O 'O -O B*O4,O O5#O P1"+P, P  P,%P1+P,,P1c^^# U(d/$[U[U55nUVs/sHJn[R"UR S5SSUR S5SUR S5SS9PML nn[ R "U4Sj[XaUSS956IShvN nU(Ga0n[TR5GHRup[[U55V s/sH oU;dM U PM n n U R"[XSS9V V s/sHupX;dM U PM sn n [[XuSS95V VVs/sH.un upX;dM[UURSU S -35S 9PM0 snnn 4S U0U S :XaUO0D6IShvN nUR[XSS9V V s0sHup[U [ 5(dMX_M sn n 5 UV s/sHn [U [ 5(aMU PM nn [U5[U5:XdGMS O UR#5n/n[[U55HGn X;aUR%['S X55 M'UR%UR)S 55 MI O}[TR5HdupU R"U[XuSS9VVs/sH#up[X.RSU S -35S 9PM% snn4S U0U S :XaUO0D6IShvN nMf Sn/n[XqSS9Hhunn[U[ 5(a-U=(d UnUR%UR+U55 MHUR%UR-U55 Mj [ R "U6IShvN U(dUc ['SU5$Ues snfGNs sn fs sn n fs snnn fGNHs sn n fs sn fs snnfNNL![.a[m[ R "U4SjU56IShvN U(a(['SUVs/sHnTPM Os snfsn5sSm@$eSm@ff=f7f)NrFrrrc 3># UHPupnURSUURS5=(d TR5URSS5S9v MR g7f)Nrrrr)rrrrrts rxr*RunnableSequence.abatch.. s\ +&B !!J/B4==?!::h5 " +sAAr'rrrrrr9rc3D># UHoRT5v M g7fr r)rrrs rxrrq s"O,B#4#4Q#7#7,r)r/rr$rrrIrr.r-rvrr=r2rrrr rrrr0rrr)rtrrrrrrrrrrrrrrrrr<rr%rmrs` @rxr=RunnableSequence.abatch sDI"&#f+6"  " ! * *&,jj&= $!'F!3%+ZZ %;# "   ELNN +.%wu+  E ? V " ;=!%.tzz%:MG$)W#6&#6aCT:T#6#&$(;;+.nU*S*S 9 *S4= #L% H4  4? %L ..%( ( ( O$;NF33! !g   ? 2&   , W" 6 ) .."O,"OP P P N,?1Q,?@@   s#QAN;8+Q#O$Q);O-$ O1O7O- O !O 'O-O#O6O- OO-*O O  O-O8O>O-B*O-*O#/O-O)O-BQO+"QO-O-+Q- Q7 Q PQ . P; : Q QQ Q  QQc +B# UR/URQURPn[SU5n[ U5HOupx[ X2R SUS-35S9nUS:XaUR"Xc40UD6nM>URXc5nMQ UShvN gN7f)Nr8rrrr)rwrrrr-r2rrs) rtrrrrrvfinal_pipelineidxrs rx _transformRunnableSequence._transform s5dkk549950&9"5)IC!"7"7)C!G98M"NFax!%!Q&!Q!%!G*"!!sBBBBc V# UR/URQURPn[SU5n[ U5HPupx[ UUR SUS-35S9nUS:XaUR"Xc40UD6nM?URXc5nMR UShvN n U 7v MN g7f)Nr:rrrr)rwrrrr-r2rrw) rtrrrrrvrrrrvs rx _atransformRunnableSequence._atransform s5dkk54995 5v>"5)IC!%//)C!G90EFFax!%!R6!R!%!H*+ &L Ns*BB)B'B%B'B)%B''B)c +# UR"UUR[X"=(d 0RS5=(d URS940UD6ShvN gN7fNr)r)rrr2rrhrs rxrsRunnableSequence.transform sS55  OO >$w-B6BBB ",*,c # UR"UUR[X"=(d 0RS5=(d URS940UD6ShvN nU7v MN g7fr)r rr2rrhrtrrrrs rxrwRunnableSequence.atransform sc >>     # T7v g7fr rrsrx input_aiter-RunnableSequence.astream..input_aiter K rr?rwrtrrrrrs ` rxrRRunnableSequence.astream 6  ??;=&KFK %K K"9757979r) rv RunnableLikerhrgrwRunnable[Any, Any] | Nonerzlist[Runnable[Any, Any]] | Nonerr rNonerz list[str])rrrr)rrr rrr rrr!r#r%rr9rrrr*rr:r&) rr=rrTrr)rrrr8) rr?rrSrr)rrrr:r<r7r>r9)'rprrIrJrKrLr rrr classmethodr"rrMrvrrrmrrrrrrrrrrrrr=rrrsrOrwrRrOrnros@rxrr sQl /',T'BF $B0 .  +/26*. ' ' ' ) ' 0 ' ( '  ' ' R 3355  $L $$ $$ 5 5.2 6+ 6  6 6     B   L  ,  4 L  -  4<@!*!*$9!*LO!* !*!*F)-%*%*&%* %*  %*%*N@D}" #( }"}"=}"  }"  }" }"}"~@D@" #( @"@"=@"  @"  @" @"@"D""0" "  "  ".$5     2)-    &        )-CC&C C  CC)- # &     )-  &     r{rc^\rSrSr%SrS\S'SSU4Sjjjr\\SSj55r \\SSj55r \ "S S 9r \SSS .S U4S jjjj5r \\S!S j55r\SS"U4Sjjj5r\SS"Sjj5r\\S#Sj55r\SS$Sjj5r\S%Sj5r\SS&Sjj5r\SS'Sjj5rS(Sjr\SS)Sjj5r\SS*Sjj5rS+Sjr\SS,Sjj5r\SS-Sjj5rSrU=r$).ri aRunnable that runs a mapping of `Runnable`s in parallel. Returns a mapping of their outputs. `RunnableParallel` is one of the two main composition primitives, alongside `RunnableSequence`. It invokes `Runnable`s concurrently, providing the same input to each. A `RunnableParallel` can be instantiated directly or by using a dict literal within a sequence. Here is a simple example that uses functions to illustrate the use of `RunnableParallel`: ```python from langchain_core.runnables import RunnableLambda def add_one(x: int) -> int: return x + 1 def mul_two(x: int) -> int: return x * 2 def mul_three(x: int) -> int: return x * 3 runnable_1 = RunnableLambda(add_one) runnable_2 = RunnableLambda(mul_two) runnable_3 = RunnableLambda(mul_three) sequence = runnable_1 | { # this dict is coerced to a RunnableParallel "mul_two": runnable_2, "mul_three": runnable_3, } # Or equivalently: # sequence = runnable_1 | RunnableParallel( # {"mul_two": runnable_2, "mul_three": runnable_3} # ) # Also equivalently: # sequence = runnable_1 | RunnableParallel( # mul_two=runnable_2, # mul_three=runnable_3, # ) sequence.invoke(1) await sequence.ainvoke(1) sequence.batch([1, 2, 3]) await sequence.abatch([1, 2, 3]) ``` `RunnableParallel` makes it easy to run `Runnable`s in parallel. In the below example, we simultaneously stream output from two different `Runnable` objects: ```python from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableParallel from langchain_openai import ChatOpenAI model = ChatOpenAI() joke_chain = ( ChatPromptTemplate.from_template("tell me a joke about {topic}") | model ) poem_chain = ( ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | model ) runnable = RunnableParallel(joke=joke_chain, poem=poem_chain) # Display stream output = {key: "" for key, _ in runnable.output_schema()} for chunk in runnable.stream({"topic": "bear"}): for key in chunk: output[key] = output[key] + chunk[key].content print(output) # noqa: T201 ``` z"Mapping[str, Runnable[Input, Any]]ruNc >Ub0UEO0nURU5 [TU] UR5VVs0sHupEU[ U5_M snnS9 gs snnf)zsCreate a `RunnableParallel`. Args: steps__: The steps to include. **kwargs: Additional steps to include. N)ru)rrTrrr)rtrurmergedrarros rxrRunnableParallel.__init__=s[(!( 3G f >DllnMnFCS,Q//nM  MsA cgrrrs rxr#RunnableParallel.is_lc_serializableWrr{c /SQ$rrrs rxr!RunnableParallel.get_lc_namespace]rr{Trric>U=(d@ UR=(d- SSRURR55S3n[TU]XS9$)zGet the name of the `Runnable`. Args: suffix: The suffix to use. name: The name to use. Returns: The name of the `Runnable`. zRunnableParallel<,>ri)rhrrurrTryrtrurhros rxryRunnableParallel.get_nameksMXtyyX&7ARARAT8U7VVW$Xw22r{cURR5H"nUR(dMURs $ [$r)rurrr)rtrs rxrRunnableParallel.InputTypezs5LL'')D~~~~~%* r{c >^[U4SjURR555(GaURR5HYnURT5RnUR S5nUcM4UR [:wdMJ[TU] T5s $ [URS5URR5VVVs0sHVnURT5RR5H%upVUS:wdM XVR UR4_M' MX snnnS9$[TU] T5$s snnnf)rc3># UH7nURT5R5RSS5S:Hv M9 g7f)r^objectN)rrr)rrrs rxr4RunnableParallel.get_input_schema..sC +   v & 8 8 : > >vx P *s?Arr9__root__r) allrurrr_rrrrTrRryrr)rtrrr\ root_fieldryrzros ` rxr!RunnableParallel.get_input_schemas,  \\((*    ++-..v6CC#ZZ/ )j.C.Cs.J 73F;; .# g&!% 3 3 5# 5 $ 5 5f = J J P P RJ1A aii00 R 5# w'//#s 9E EcURR5VVs0sHup#X#RS4_M nnn[UR S5US9$s snnf)r.r:r)rurrrRry)rtrryrzr\s rxr"RunnableParallel.get_output_schemasQ6:\\5G5G5IJ5ITQ!llC((5IJt}}X6&QQKsAcV[SURR555$)rc3J# UHoRHo"v M M g7fr r)rrrs rxr0RunnableParallel.config_specs..s!' 2T?P?PtD?PD2!#)rCrurrs rxrRunnableParallel.config_specss+'' !\\002'   r{c@SSKJn U"5nURURU55nURUR U55nUR R 5HnUR5nUR5 UR5 U(dURXE5 MMURU5upU(dSUS3n [U 5eU (dSUS3n [U 5eURXH5 URX5 M U$)rrrYrr has no last node) rrZrrrrurrrrrrr`) rtrrZrrrrrrstep_last_noders rxrRunnableParallel.get_graphs 9^^D$9$9&$AB nnT%;%;F%CD LL'')D)J  & & (  % % 'z727,,z2J/&%dV+=>C$S/)%%dV+<=C$S/)z;~;*" r{crSRSURR555nSU-S-$)Nz, c3f# UH'upUS[[U5SU-S-53v M) g7f)z: z N)rDr)rryrzs rxr,RunnableParallel.__repr__..s9$ ,c,T!WdQhoFG H,/1z{ z })rrur)rt map_for_reprs rxrRunnableParallel.__repr__s<||$  **,$  %--r{c ^[U5n[R"URS5SSURS5SURS5SS9nUR SUURS5=(d UR 5UR SS5S9mS U4S jjn[UR5n[U5nUR5VV s/sHupURXYXU5PM n nn [XjSS 9VV s0sHupXR5_M n nn SSS5 TRW 5 U $s sn nfs sn nf!,(df  N-=f![an TR!U 5 eSn A ff=f) NrFrrrrrrc>[UTRSU35S9n[U5nURURUU5sSSS5 $!,(df  g=f)Nmap:key:r)r2rr4rrrrrrarrrs rx _invoke_step-RunnableParallel.invoke.._invoke_step sX(%//(3%0@AL $L1W{{KK 211s A Ar' rrrr9rr)rarrr)r,r%rrrryr0rrur0rr/r.r1rrr)rtrrrrrBrvrrarr3r5rvrrs @rxrRunnableParallel.invokes v&*44"(**["9 #ZZ/!'J!7 '55  J':4==?::h- 6  & 05 ?M TW     &E(0H&+[[]%2 OOLsK%2 (+5%'H'H ('H 1  $ $V ,M 10   & &q )  sN/ E)E#E EE.E1E) E E&"E)) F 3FF c Z^^^ ^ # [T5m[T5nURSTTRS5=(d UR 5TR SS5S9IShvN m SU 4Sjjm [ UR5n[R"U UU4SjUR556IShvN n[ [XVSS95nT RU5IShvN U$NN4N ![a nT RU5IShvN eSnAff=f7f) Nrrrc># [UTRSU35S9n[U5n[UR X5USS9IShvN sSSS5 $N !,(df  g=f7f)Nr@rTr)r2rr4r=rrAs rx _ainvoke_step/RunnableParallel.ainvoke.._ainvoke_stepAsh(%//(3%0@AL$L1W.LL6T2121s.(A*A A A A*A A'#A*c3>># UHupT"UTTU5v M g7fr r)rrarrHrrs rxr+RunnableParallel.ainvoke..Rs4 &3 " &3sFr'rD)r,r-rrryr0rrurIrrr.rrr) rtrrrrrvresultsrvrrHrs `` @@rxrRunnableParallel.ainvoke/s0v&@H,;;  J':4==?::h- <   & 05 ?M TW    &E#NN &+[[]   G#eU;C:C>D+1C<2D+:C><D+> D(D#DD##D((D+c## [UR5n[[U[ U5[ R "5S95n[U5nUR5VVs/sH@upxUURUR5[X2RSU35S954PMB n nnU V V s0sHupUR[U 5X4_M n n n U (an[U [ S9upU HOnU RU5up[#XR%505nUv U U 4XR[U 5'MQ U (aMnSSS5 gs snnfs sn n f![&a M~f=f!,(df  g=f7f)Nlockr@rr))rrurrQr threadingLockr0rrsr0r2rr/rrrr5r1r)rtrrrrv input_copiesrrhrnamed_generators step_name generatorr3completed_futuresrmr5rs rxrRunnableParallel._transformfs~T\\"GFCJY^^=MNO $V ,#(++- #0JDNN$((*$".C.ChtfDU.V#0  -=,<(Ii092HH,< '+G'Q$!/F-4[[-@*Y +Y ,H I# %%Ei @A 0'-- , $)C- ,saAF E8'AE.E86$E!2E8 8E' E8 F  E8' E51E84E55E88 FF c+\# UR"XRU40UD6ShvN gN7fr )rrrs rxrsRunnableParallel.transforms055 ??F .4   rc+X# UR[U/5U5ShvN gN7fr rrs rxrORunnableParallel.streams!>>$w-888s *(*c4# [UR5n[[U[ U5[ R "5S95nUR5VVs/sH@upgUURUR5[X2RSU35S954PMB nnnSSjn UV V s0sH#up[ R"U "U 55X4_M% n n n U (a[ R"U [ RS9IShvN upU HXnU RU5up[XR!505nU7v [ R"U "U 55nX4U U'MZ U (aMggs snnfs sn n fN{!["a Mf=f7f)NrOr@rc4# [U5IShvN $N7fr )r)rVs rxget_next_chunk4RunnableParallel._atransform..get_next_chunksy)) ))s r))rVrrz Output | None)rrurrPrrIrRrrwr0r2rrrrr5r1r )rtrrrrvrSrhrrTr_rUrVtaskscompleted_tasksrmtaskrnew_tasks rxrRunnableParallel._atransformsT\\"DU',,.IJ $kkm  ,  $$& *?*?(4&@Q*R ,    * )9 (8$    y 9 :Y)3LMH'0&> ##V /5  %K  s5313535c r^# SU4SjjnURU"5U5ShvN nU7v MN g7f)Nc># T7v g7fr rrsrxr-RunnableParallel.astream..input_aiterrrrrrs ` rxrRRunnableParallel.astreams1  ??;=&A %K As 7535757rr )ruzMapping[str, Runnable[Input, Any] | Callable[[Input], Any] | Mapping[str, Runnable[Input, Any] | Callable[[Input], Any]]] | NonerzkRunnable[Input, Any] | Callable[[Input], Any] | Mapping[str, Runnable[Input, Any] | Callable[[Input], Any]]rrrrrrrrrr r)rr9rrrrrr)rr9rrrr*rr)rr=rrTrr)rzIterator[AddableDict])rr=rrrrrIterator[dict[str, Any]])rr9rrrr*rrm)rr?rrSrr)rzAsyncIterator[AddableDict])rr?rrrrrAsyncIterator[dict[str, Any]])rr9rrrr*rrn)rprrIrJrKrLrrr"rrrrmryrMrrrrrrrrrrsrOrrwrRrOrnros@rxrr sQf0/  F    4  33 $L 3 3 3 3  0 0 0D.2 R+ R  R R     $$L..<@99$99LO9 99v)-44&4 4  44l--0- -  -^)-  &   "   )-99&9 9 " 991$151 1 $ 1f)- # &   '  )-  &   '  r{rc\rSrSrSrSSS.SSjjjr\\SSj55r\SSSjj5r \\SSj55r \SSS jj5r \SS j5r Sr \SS j5r\SSS jj5r\SSS jj5r\SSSjj5r\SSSjj5r\SSSjj5r\SSSjj5rSrg)RunnableGeneratoriu, `Runnable` that runs a generator function. `RunnableGenerator`s can be instantiated directly or by using a generator within a sequence. `RunnableGenerator`s can be used to implement custom behavior, such as custom output parsers, while preserving streaming capabilities. Given a generator function with a signature `Iterator[A] -> Iterator[B]`, wrapping it in a `RunnableGenerator` allows it to emit output chunks as soon as they are streamed in from the previous step. !!! note If a generator function has a `signature A -> Iterator[B]`, such that it requires its input from the previous step to be completed before emitting chunks (e.g., most LLMs need the entire prompt available to start generating), it can instead be wrapped in a `RunnableLambda`. Here is an example to show the basic mechanics of a `RunnableGenerator`: ```python from typing import Any, AsyncIterator, Iterator from langchain_core.runnables import RunnableGenerator def gen(input: Iterator[Any]) -> Iterator[str]: for token in ["Have", " a", " nice", " day"]: yield token runnable = RunnableGenerator(gen) runnable.invoke(None) # "Have a nice day" list(runnable.stream(None)) # ["Have", " a", " nice", " day"] runnable.batch([None, None]) # ["Have a nice day", "Have a nice day"] # Async version: async def agen(input: AsyncIterator[Any]) -> AsyncIterator[str]: for token in ["Have", " a", " nice", " day"]: yield token runnable = RunnableGenerator(agen) await runnable.ainvoke(None) # "Have a nice day" [p async for p in runnable.astream(None)] # ["Have", " a", " nice", " day"] ``` `RunnableGenerator` makes it easy to implement custom behavior within a streaming context. Below we show an example: ```python from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableGenerator, RunnableLambda from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser model = ChatOpenAI() chant_chain = ( ChatPromptTemplate.from_template("Give me a 3 word chant about {topic}") | model | StrOutputParser() ) def character_generator(input: Iterator[str]) -> Iterator[str]: for token in input: if "," in token or "." in token: yield "👏" + token else: yield token runnable = chant_chain | character_generator assert type(runnable.last) is RunnableGenerator "".join(runnable.stream({"topic": "waste"})) # Reduce👏, Reuse👏, Recycle👏. # Note that RunnableLambda can be used to delay streaming of one step in a # sequence until the previous step is finished: def reverse_generator(input: str) -> Iterator[str]: # Yield characters of input in reverse order. for character in input[::-1]: yield character runnable = chant_chain | RunnableLambda(reverse_generator) "".join(runnable.stream({"topic": "waste"})) # ".elcycer ,esuer ,ecudeR" ``` Nric*UbX lUn[U5(a XlUnO=[R"U5(a XlUnOS[ U53n[ U5eU=(d URUlg![a SUlgf=f)zInitialize a `RunnableGenerator`. Args: transform: The transform function. atransform: The async transform function. name: The name of the `Runnable`. Raises: TypeError: If the transform is not a generator function. NzTExpected a generator function type for `transform`.Instead got an unsupported type: rp) rrFrisgeneratorfunctionrr^rrprhAttributeError)rtrsrwrh func_for_namers rxrRunnableGenerator.__init__Vs(  !) &0M i ( (( %M  ( ( 3 3'O%M448O3DF C.  ,6 6 6DI ,+DI ,s#A>>BBc[USS5=(d URn[R"U5Rn[ [ UR55S5nU(aGUR[RR:wa[URS[45S$[$![a [$f=fNr__args__r) getattrrr signature parametersrrrr Parameteremptyrr`rtrparams first_params rxrRunnableGenerator.InputTypest\40DD4D4D &&t,77FtFMMO4d;K{559J9J9P9PP{55zC6J1MM     sBB99 C  C c6URn[USS5=(d URn[USS5n[R"U5(a,[ U[ 5(d[U[5(aU$[URS5UUS9$)Nrrr9r) rryrrrrrrrrRryrtrrrmodules rxr"RunnableGenerator.get_input_schemas NN t\40DD4D4D|T2 OOI & &y,779i00  MM' "   r{c6[USS5=(d URn[R"U5nUR[R R :wa[URS[45S$[$![a [s$f=frw) ryrrrzreturn_annotation Signaturer}rr`rtrsigs rxrRunnableGenerator.OutputTypest\40DD4D4D ##D)C((G,=,=,C,CC--zC6B1E    J sAB?BBBc6URn[USS5=(d URn[USS5n[R"U5(a,[ U[ 5(d[U[5(aU$[URS5UUS9$)Nrrr:r) rryrrrrrrrrRryrs rxr#RunnableGenerator.get_output_schemasOO t\40DD4D4D|T2 OOI & &y,779i00  MM( #   r{c[U[5(aw[US5(a*[US5(aURUR:H$[US5(a*[US5(aURUR:H$gg)NrrF)rrprnrrrs rx__eq__RunnableGenerator.__eq__st e. / /t\**wul/K/K%*:*:::t]++}0M0M''5+<+<<<r{c"SURS3$)NzRunnableGenerator()rirs rxrRunnableGenerator.__repr__s#DII;a00r{c [US5(dU<S3n[U5eUR"UURU4SS0UD6$)Nrz only supports async methods.rT)rnrjrrrtrrrrs rxrsRunnableGenerator.transforms]t\**H9:C%c* *11  OO       r{c >UR"[U/5U40UD6$r rrs rxrORunnableGenerator.stream ~~dE7mV>v>>r{c dSnUR"X40UD6H nUcUOXE-nM [SU5$Nr:)rOrrtrrrrrrvs rxrRunnableGenerator.invokes; $kk%:6:F#mFE;He$$r{c [US5(dU<S3n[U5eUR"XRU4SS0UD6$)Nrz only supports sync methods.rT)rnrjr rrs rxrwRunnableGenerator.atransformsWt]++H89C%c* *22 ##V ;? CI  r{c F^SU4SjjnUR"U"5U40UD6$)Nc># T7v g7fr rrsrxr.RunnableGenerator.astream..input_aiterrrrr)rtrrrrs ` rxrRRunnableGenerator.astreams" {}f???r{cz# SnUR"X40UD6ShvN nUcUOXE-nMN [SU5$7fr)rRrrs rxrRunnableGenerator.ainvokesH $ LLA&A @&#mFE @AHe$$s;.,. ;. ;)rrrhr )rszgCallable[[Iterator[Input]], Iterator[Output]] | Callable[[AsyncIterator[Input]], AsyncIterator[Output]]rwz>Callable[[AsyncIterator[Input]], AsyncIterator[Output]] | Nonerhrgrrrlrrr&rr)rrr=rrrrrr8)rr9rrrrrr8r%rr?rrrrrr:)rr9rrrrrr:)rprrIrJrKrrMr"rrrrr__hash__rrsrOrrwrRrrOrr{rxrprpsWY@ (, (,B(, (,(, (,T     2   .2 +   4H 11)-  &      ")-??&? ?  ??<@%%$9%LO% %%)-  #  &        )- @ @& @ @  @ @<@%%$9%LO% %%r{rpc^\rSrSrSrSSSjjr\\SSj55r\SSU4Sjjj5r \\SSj55r \SSSjj5r \ RSS j5r\\SS j55r\SS U4S jjj5r\S!S j5rSrS"S jrS#SjrS$Sjr\SS%Sjj5r\SS%Sjj5rS&Sjr\SS'Sjj5r\SS(Sjj5rS)Sjr\SS*Sjj5r\SS+Sjj5rSrU=r$),RunnableLambdai)a`RunnableLambda` converts a python callable into a `Runnable`. Wrapping a callable in a `RunnableLambda` makes the callable usable within either a sync or async context. `RunnableLambda` can be composed as any other `Runnable` and provides seamless integration with LangChain tracing. `RunnableLambda` is best suited for code that does not need to support streaming. If you need to support streaming (i.e., be able to operate on chunks of inputs and yield chunks of outputs), use `RunnableGenerator` instead. Note that if a `RunnableLambda` returns an instance of `Runnable`, that instance is invoked (or streamed) during execution. Examples: ```python # This is a RunnableLambda from langchain_core.runnables import RunnableLambda def add_one(x: int) -> int: return x + 1 runnable = RunnableLambda(add_one) runnable.invoke(1) # returns 2 runnable.batch([1, 2, 3]) # returns [2, 3, 4] # Async is supported by default by delegating to the sync implementation await runnable.ainvoke(1) # returns 2 await runnable.abatch([1, 2, 3]) # returns [2, 3, 4] # Alternatively, can provide both synd and sync implementations async def add_one_async(x: int) -> int: return x + 1 runnable = RunnableLambda(add_one, afunc=add_one_async) runnable.invoke(1) # Uses add_one await runnable.ainvoke(1) # Uses add_one_async ``` NcUbX lUn[U5(d[U5(aUb Sn[U5eXlUnO=[ U5(a[ SU5UlUnOS[U53n[U5eUbX0lO!URS:waURUlSUl g![a Nf=f)aCreate a `RunnableLambda` from a callable, and async callable or both. Accepts both sync and async variants to allow providing efficient implementations for sync and async execution. Args: func: Either sync or async callable afunc: An async callable that takes an input and returns an output. name: The name of the `Runnable`. Raises: TypeError: If the `func` is not a callable type. TypeError: If both `func` and `afunc` are provided. NzFunc was provided as a coroutine function, but afunc was also provided. If providing both, func should be a regular function to avoid ambiguity.Callable[[Input], Output]zEExpected a callable type for `func`.Instead got an unsupported type: z) afuncrErFrcallablerrr^rhrprs_repr)rtrrrhrtrs rxrRunnableLambda.__init__YsR  J&+M T " "&8&>&> 3  n$J M d^^8$?DI M448J<A C.   '':5)22 "&    s B6 !B66 CCcv[USS5=(d URn[R"U5Rn[ [ UR55S5nU(a4UR[RR:wa UR$[$![a [$f=f)z)The type of the input to this `Runnable`.rN) ryrrrzr{rrrrr|r}r`rr~s rxrRunnableLambda.InputTypestVT*8djj &&t,77FtFMMO4d;K{559J9J9P9PP"---     sA=B&& B87B8c>[USS5=(d URn[U[5(a[ U5R SS5SSR S5n[SU55(a4UVs0sHoDSS[S 4_M nn[URS 5US 9$[US S5n[URS 5[[US 9$UR[:wa[TU]9U5$[U5=n(a3[URS 5[ R#U[S 45S 9$[TU]9U5$s snf)zThe Pydantic schema for the input to this `Runnable`. Args: config: The config to use. Returns: The input schema for this `Runnable`. rNzoperator.itemgetter(rr|z, c3f# UH'oSS:H=(a USS:H=(a US:gv M) g7f)r'r|z''Nr)rres rxr2RunnableLambda.get_input_schema..s3PUQ3C48s?Ctt|CPUr;r.r9rrr)ryrrrrreplacesplitr)rrRryrrrTrr@rfromkeys) rtrrrrer\r dict_keysros rxrRunnableLambda.get_input_schemasMtVT*8djj dJ ' 'I%%&CCUTq*sCj0UC&t}}W'=QWXXT<6F" g&#Y#   >>S 7+F3 38> >9 >" g&"&-- C:"F  w'//-Ds?Ec[USS5=(d URn[R"U5nUR[R R :wax[URSS5[RR[RR1;a[URS[45S$UR$[$![a [$f=f)zzThe type of the output of this `Runnable` as a type annotation. Returns: The type of the output of this `Runnable`. rN __origin__rxr) ryrrrzrrr} collectionsabcr rrr`rs rxrRunnableLambda.OutputTypestVT*8djj ##D)C$$(9(9(?(??300,EOO,,OO11J#3#8#8*sfMaPP,,,@     sB)C C C10C1c6URn[USS5=(d URn[USS5n[R"U5(a,[ U[ 5(d[U[5(aU$[URS5UUS9$)Nrrr:r) rryrrrrrrrrRryrs rxr RunnableLambda.get_output_schemasOO tVT*8djj|T2 OOI & &y,779i00  MM( #   r{c[US5(a[UR5nO)[US5(a[UR5nO/n/nUHhn[ U[ 5(aUR U5 M+[ [USS5[ 5(dMMUR UR5 Mj U$)zThe dependencies of this `Runnable`. Returns: The dependencies of this `Runnable`. If the function has nonlocal variables that are `Runnable`s, they are considered dependencies. rr__self__N) rnrArrrrerryr)rtobjectsdepsobjs rxrRunnableLambda.depss 4 ,TYY7G T7 # #,TZZ8GG!C#x(( C GCT:HEE CLL)   r{c:[SUR55$)Nc3J# UHoRHo"v M M g7fr r0)rdeprs rxr.RunnableLambda.config_specs..,s ' %S2B2B$D2BDIr2)rCrrs rxrRunnableLambda.config_specs)s"''  II'   r{c\>UR=n(GaSSKJn U"5nURUR U55nURUR U55nUHnUR 5nUR5 UR5 U(dURXV5 MMURU5upU (dSUS3n [U 5eU (dSUS3n [U 5eURXY5 URX5 M U$[T U]U5nU$)NrrYrrr5) rrrZrrrrrrrrr`rT) rtrrrZrrrr dep_graphdep_first_node dep_last_noderros rxrRunnableLambda.get_graph0s99 4  <GE(=(=f(EFJ..)?)?)GHKMMO ))+((* NN:;49LL4K1N) )#.@A(o-( )#.?@(o-NN:>NN=>& G%f-E r{c[U[5(aw[US5(a*[US5(aURUR:H$[US5(a*[US5(aURUR:H$gg)NrrF)rrrnrrrs rxrRunnableLambda.__eq__Nsm e^ , ,tV$$)?)?yyEJJ..tW%%'%*A*AzzU[[00r{cURc[US5(aU[UR[5(a6S[ UR5[ S5SS3UlUR$[US5(a3S[UR5=(d SS3UlUR$[US5(a3S[UR5=(d SS3UlUR$S UlUR$) z2Return a string representation of this `Runnable`.NrzRunnableLambda(z operator.rz...rzRunnableLambda(afunc=zRunnableLambda(...)) rrnrrrrrrBrrs rxrRunnableLambda.__repr__Zs :: tV$$DIIz)J)J.s499~c+>N>P/Q.RRST zzv&&./@/K/Tu.UUVW zz w''+,=djj,I,RU+SSTU zz3 zzr{c [R"UR5(a9Sn[[ SUR5UUU40UD6HnUcUnM XV-nM O[URXU40UD6n[ U[5(aKUSnUS::aSUSUS3n[U5eURU[UUR5US- S95n[ S U5$![ a UnMf=f) N#Callable[[Input], Iterator[Output]]recursion_limitr&Recursion limit reached when invoking with input .rrrr:) rrrrr+rrrreRecursionErrorrr2r) rtrrrrrvrrrs rx_invokeRunnableLambda._invokeis"  & &tyy 1 1$(F5:DIIF   >"F'!'2 6;:@F fh ' '$%67O!#Sn[[STR5UUUR540UD6HnUcUnM XE-nM [SU5$![a UnM.f=f)Nrr:)r+rrget_syncr)r9rrrrvrrts rxr%RunnableLambda._ainvoke..funcs -1F!=BDIIN#,,. " ! "">%*F/)/" &11$-/)./sA A$#A$cP>[TRXUR540UD6$r r+rr)r9rrrrts rxrrs. 8 5+2F2F2HLRr{c@># [TT/UQ70UD6IShvN $N7fr r3r}rrrs rxf"RunnableLambda._ainvoke..f#,VTKDKFKKKK zAsyncGenerator[Any, Any]r r:rrrrrrr r9r9rrSrr)rrrr:r}rrrrr)rnrrrrrrrFrOrr*rrrerrr2r) rtr9rrrrrrvrOrrrrs ` ` @rx_ainvokeRunnableLambda._ainvokesA 4 ! !JJE**499552 2!@2+2" 2  222 !@+"   4[ L LE e $ $$(F.1Z/#  !    #'+$ +%~!&+%+^F9Z' GMF fh ' '$%67O!#>)335$3a$7FHf%%W  + )+%*F+$      0sCG F G F1FFF"F1+F /F11G G A'G5G 6GF FF1FF1 G)F,*G1G7F: 8 GG Gc [US5(a(UR"URU[U540UD6$Sn[ U5e)aInvoke this `Runnable` synchronously. Args: input: The input to this `Runnable`. config: The config to use. **kwargs: Additional keyword arguments. Returns: The output of this `Runnable`. Raises: TypeError: If the `Runnable` is a coroutine function. rzGCannot invoke a coroutine function synchronously.Use `ainvoke` instead.)rnrrr,rrs rxrRunnableLambda.invokesR* 4 )) f%   Xnr{cn# UR"URU[U540UD6IShvN $N7f)zInvoke this `Runnable` asynchronously. Args: input: The input to this `Runnable`. config: The config to use. **kwargs: Additional keyword arguments. Returns: The output of this `Runnable`. N)rrr,rs rxrRunnableLambda.ainvokes@$,, MM  & !      s ,535c +# SnUHnU(dUnSnMWU-nM [R"UR5(a2Sn[ URWX240UD6Hn U v UcU nMX-nM O[ URWX240UD6n[ U[ 5(aWUSn U S::aSUSUS3n [U 5eURU[UUR5U S- S 95Hn U v M g[R"UR5(d[S U5v gg![a UnGM:f=f![a U nMf=f7f) NFTrrrrrrrr:) rrrrrr+rrerrOr2rr) rtchunksrrrrprqrrrvrrrs rxrRunnableLambda._transform2s F ! $ #!FNE  & &tyy 1 1$(F5 5&9? >"F'!'2 5&9?F fh ' '$%67O!#E1 E=EEE EEEEc+# [US5(a1UR"UUR[U540UD6ShvN gSn[ U5eN7f)NrzGCannot stream a coroutine function synchronously.Use `astream` instead.)rnrrr,rrs rxrsRunnableLambda.transformpsc 4 99f%   ) C.  s=AAAc >UR"[U/5U40UD6$r rrs rxrORunnableLambda.streamrr{c ^^^# SnUShvN nU(dUnSnMWU-nMN![a UnM/f=f [TS5(a TRnO][R"TR 5(a Sn [U 5eSU4Sjjm[ T5SUU4Sjj5n U n[U5(aWSn [S[[S U5WTU40UD65ShvN n U 7v U cU n MX-n M![a U n M,f=f O"[[S U5WTU40UD6IShvN n [U [5(a^TS n U S ::aS TS US3n [U 5eU RU[TUR5U S- S95ShvN n U 7v M g[U5(d[SU 57v gg7f)NFTrzMCannot stream from a generator function asynchronously.Use .stream() instead.r:cP>[TRXUR540UD6$r r)rrrrrts rxr(RunnableLambda._atransform..funcs. 4IIv{/C/C/EIOr{c@># [TT/UQ70UD6IShvN $N7fr rrs rxr%RunnableLambda._atransform..frrr:r rrrrrrr) rr9rrSrr)rrrr:r)rrnrrrrrrrFrr*rrerrRr2r)rtrrrrrprqrrrrrrvrrrs` ` @rxrRunnableLambda._atransformsL " #& ! $ #!FNE #!#"E## 4 ! !JJE**49955- n$  < '      4[ L LE e $ $$(F#'-U+     ' 'e >"F'!'$'!&'! &9Z'   F fh ' '$%67O!## T7v g7fr rrsrxr+RunnableLambda.astream..input_aiterrrrrrs ` rxrRRunnableLambda.astreamr r )rrrrhrG)raACallable[[Input], Iterator[Output]] | Callable[[Input], Runnable[Input, Output]] | Callable[[Input], Output] | Callable[[Input, RunnableConfig], Output] | Callable[[Input, CallbackManagerForChainRun], Output] | Callable[[Input, CallbackManagerForChainRun, RunnableConfig], Output] | Callable[[Input], Awaitable[Output]] | Callable[[Input], AsyncIterator[Output]] | Callable[[Input, RunnableConfig], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun, RunnableConfig], Awaitable[Output]]ra-Callable[[Input], Awaitable[Output]] | Callable[[Input], AsyncIterator[Output]] | Callable[[Input, RunnableConfig], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun, RunnableConfig], Awaitable[Output]] | Nonerhrgrrrlr r)rzlist[Runnable]rr rr) rr9rrTrr)rrrr:rr) rr=rrTrr)rrrr8r<r7) rr?rrSrr)rrrr:r>r9) rprrIrJrKrrMr"rrrrrcached_propertyrrrrrrrrrrrrsrOrrwrRrOrnros@rxrr)s-J/I&  I&I&./I&0 1I&V   )0)0)0V ,.2 +   4.   :H +&+&0+& +&  +&  +&Zb&b&5b& b&  b&  b&H)-&   <)-  &      0<)<)0<) <)  <)  <)|)-!!&! !  !!()-??&? ?  ??a)$a)5a) a)  a)  a)F)- # &     )-  &     r{rc\rSrSr%SrS\S'\"SS9r\\ SSj55r \ SSS jj5r \\ SS j55r \ SSS jj5r \\ SS j55r\ SSS jj5r\\ SSj55r\\ SSj55rSSjr\ SSSjj5rS Sjr\ SSSjj5r\ SS!Sjj5rSrg)"RunnableEachBaseizRunnableEachBase class. `Runnable` that calls another `Runnable` for each element of the input sequence. Use only if creating a new `RunnableEach` subclass with different `__init__` args. See documentation for `RunnableEach` for more details. rArzTrc<[URR$r )rrzrrs rxrRunnableEachBase.InputType sDJJ(())r{Nc[URS5[URR U5S4UR R S9$)Nr9r)rRryrrzrrorrs rxr!RunnableEachBase.get_input_schema%sI MM' "TZZ008911  r{c<[URR$r )rrzrrs rxrRunnableEachBase.OutputType7sDJJ))**r{cURRU5n[URS5[UUR R S9$)Nr:r)rzrrRryrror)rtrrs rxr"RunnableEachBase.get_output_schema<sG--f5 MM( #f11  r{c.URR$r rzrrs rxrRunnableEachBase.config_specsNzz&&&r{c8URRU5$r )rzrrs rxrRunnableEachBase.get_graphSszz##F++r{cgrrrs rxr#RunnableEachBase.is_lc_serializableWrr{c /SQ$rrrs rxr!RunnableEachBase.get_lc_namespace]rr{c UVs/sHn[X2R5S9PM nnURR"X40UD6$s snfNr)r2rrzrrtrrrrrmrs rxrRunnableEachBase._invokegsQNT MSL+@+@+B CV  zz:6:: s!Ac >UR"URX40UD6$r )rrrs rxrRunnableEachBase.invokess %%dllELVLLr{c# UVs/sHn[X2R5S9PM nnURR"X40UD6IShvN $s snfN 7fr )r2rrzr=r!s rxrRunnableEachBase._ainvokeys]NT MSL+@+@+B CV  ZZ&&vA&AAA BsA!A"A A AcZ# UR"URX40UD6IShvN $N7fr )rrrs rxrRunnableEachBase.ainvokes),,T]]ETVTTTTs "+)+c P# SSjn[S5HnU"S57v M g7f)Nc[U5er )rjrs rx_error_stream_event._error_stream_events %g. .r{rz1RunnableEach does not support astream_events yet.)rrrr\)r)rtrrrr+rms rxrmRunnableEachBase.astream_eventss* /qA%C s$&rrlr r)rztype[list[Output]]rr rr) rr'rrTrr)rrrr)rr'rrrrrr) rr'rrSrr)rrrrrr9rrrr*rr;)rprrIrJrKrLrrmrMr"rrrrrrrrrrrrrrmrOrr{rxr r s   #" $L **  " ++.2 +   " '',,  33 ; ;0 ; ;  ;  ;BFM M*?MRUM MM B B5 B B  B  BBFU U*?URUU UU )-  &   $  r{r c^\rSrSrSr\S SS.S U4Sjjjj5r\SSj5r\S SSjj5r\SSSS.SS jj5r SSSS.SS jjr S r U=r $)riaRunnableEach class. `Runnable` that calls another `Runnable` for each element of the input sequence. It allows you to call multiple inputs with the bounded `Runnable`. `RunnableEach` makes it easy to run multiple inputs for the `Runnable`. In the below example, we associate and run three inputs with a `Runnable`: ```python from langchain_core.runnables.base import RunnableEach from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt = ChatPromptTemplate.from_template("Tell me a short joke about {topic}") model = ChatOpenAI() output_parser = StrOutputParser() runnable = prompt | model | output_parser runnable_each = RunnableEach(bound=runnable) output = runnable_each.invoke([{'topic':'Computer Science'}, {'topic':'Art'}, {'topic':'Biology'}]) print(output) # noqa: T201 ``` Nric>U=(d1 UR=(d SURR5S3n[TU] XS9$)Nz RunnableEachAL $LL  # #  # #  %# # P*.'+)-  ' %  '  %   r{rc^\rSrSr%SrS\S'\"\S9rS\S'\"\ S9r S\S '\"\ S9r S \S 'S r S \S'S rS \S'\"SS9rS S S S S S.S1U4Sjjjr\S2S S.S3Sjjj5r\\S4Sj55r\\S5Sj55r\S2S6U4Sjjj5r\S2S6U4Sjjj5r\\S7Sj55r\S2S8Sjj5r\\S9Sj55r\\S:Sj55rS;Sjr\S2SS%jjj5r$\#S2S?S&jj5r$\S2S!S".S@S'jjj5r$\#S2S!S".SAS(jjj5r%\#S2SBS)jj5r%\S2S!S".SCS*jjj5r%\S2SDS+jj5r&\S2SES,jj5r'\S2SFS-jj5r(\S2SGS.jj5r)\S2SHS/jj5r*S0r+U=r,$)IRunnableBindingBaseiz`Runnable` that delegates calls to another `Runnable` with a set of `**kwargs`. Use only if creating a new `RunnableBinding` subclass with different `__init__` args. See documentation for `RunnableBinding` for more details. rArzrzMapping[str, Any]rr)rz0list[Callable[[RunnableConfig], RunnableConfig]]rNr*rrTr)rrrrrc >[TU]"SUU=(d 0U=(d 0U=(d /UUS.UD6 U=(d 0Ulg)a<Create a `RunnableBinding` from a `Runnable` and kwargs. Args: bound: The underlying `Runnable` that this `Runnable` delegates calls to. kwargs: optional kwargs to pass to the underlying `Runnable`, when running the underlying `Runnable` (e.g., via `invoke`, `batch`, `transform`, or `stream` or async variants) config: optional config to bind to the underlying `Runnable`. config_factories: optional list of config factories to apply to the config before binding to the underlying `Runnable`. custom_input_type: Specify to override the input type of the underlying `Runnable` with a custom type. custom_output_type: Specify to override the output type of the underlying `Runnable` with a custom type. **other_kwargs: Unpacked into the base class. rzrrrrrNr)rTrr) rtrzrrrrr other_kwargsros rxrRunnableBindingBase.__init__?sN@  URb[TU] U5$URR[ UR U55$r )rrTrrzr1rrtrros rxr$RunnableBindingBase.get_input_schemas?  ! ! -7+F3 3zz**=f+MNNr{c>URb[TU] U5$URR[ UR U55$r )rrTrrzr1rrHs rxr%RunnableBindingBase.get_output_schemasA  " " .7,V4 4zz++M$++v,NOOr{c.URR$r rrs rxr RunnableBindingBase.config_specsrr{cVURRURU55$r )rzr_merge_configsrs rxrRunnableBindingBase.get_graphs"zz##D$7$7$?@@r{cgrrrs rxr&RunnableBindingBase.is_lc_serializablerr{c /SQ$rrrs rxr$RunnableBindingBase.get_lc_namespacerr{cp^[UR/UQ76m[T/U4SjUR5Q76$)Nc32># UH o"T5v M g7fr r)rrrs rxr5RunnableBindingBase._merge_configs..s&P:OQqyy:Os)r1rr)rtrrs @rxrO"RunnableBindingBase._merge_configss2t{{5W5VQ&P$:O:O&PQQr{c xURR"UURU5400UREUED6$r )rzrrOrrs rxrRunnableBindingBase.invokesDzz      ' (''  r{c# URR"UURU5400UREUED6IShvN $N7fr )rzrrOrrs rxrRunnableBindingBase.ainvokesQZZ''     ' (''    s?AAAFrc X[U[5(a,[SUVs/sHoPRU5PM sn5nO2[ [ U55Vs/sHopRU5PM nnUR R"UU4SU00UREUED6$s snfs snfNzlist[RunnableConfig]r) rrrrOrrrzrrrtrrrrconfrrms rxrRunnableBindingBase.batchs fd # #&7=>vt$$T*v>G =B#f+vt$$T*v>G =B#f+vt$$T*v>G =B#f+UR R"UU4SU00UREUED6ShvN nU7v MUR R"UU4SU00UREUED6ShvN nU7v Ms snfs snfNS gN g7fr^) rr rrOrrrzr@r) rtrrrrr`rrmres rxr@ri>s fh ' '&7=>vt$$T*v>G =B#f+47D +D/D0D36D )D -D.D 1D DD D  D c+# URR"UURU5400UREUED6ShvN gN7fr )rzrOrOrrs rxrORunnableBindingBase.stream_sL::$$     ' (''   ?A AA c # URR"UURU5400UREUED6ShvN nU7v MN g7fr )rzrRrOrrtrrrres rxrRRunnableBindingBase.astreamlsZ**,,     ' (''  $ J   (BFNB:INJFE%)z( &* )  $L,0(,<@>B,#',#) ,# & ,#  ,#:,#<,#,# ,#,#\666      OOO .2P+P PP ''AA  33R)-    &        )-    &        @D #(   =        ,@D #(   =        ,DH+ -2 ++A+ * +  + &++DH77A7 ) 7  7 277DH #( A     2>DH0 -2 00A0 * 0  0 +00DH<<A< ) <  < 7<<DH #( A     7@)-    &        )-  &     )-  &   $  )-    &        )- # &     r{r<c\rSrSrSr\S Sj5r\S SSjj5r\SSSS.SSjj5r\SSSjj5r \S S j5r \SS j5r S r g)r|iaWrap a `Runnable` with additional functionality. A `RunnableBinding` can be thought of as a "runnable decorator" that preserves the essential features of `Runnable`; i.e., batching, streaming, and async support, while adding additional functionality. Any class that inherits from `Runnable` can be bound to a `RunnableBinding`. Runnables expose a standard set of methods for creating `RunnableBindings` or sub-classes of `RunnableBindings` (e.g., `RunnableRetry`, `RunnableWithFallbacks`) that add additional functionality. These methods include: - `bind`: Bind kwargs to pass to the underlying `Runnable` when running it. - `with_config`: Bind config to pass to the underlying `Runnable` when running it. - `with_listeners`: Bind lifecycle listeners to the underlying `Runnable`. - `with_types`: Override the input and output types of the underlying `Runnable`. - `with_retry`: Bind a retry policy to the underlying `Runnable`. - `with_fallbacks`: Bind a fallback policy to the underlying `Runnable`. Example: `bind`: Bind kwargs to pass to the underlying `Runnable` when running it. ```python # Create a Runnable binding that invokes the chat model with the # additional kwarg `stop=['-']` when running it. from langchain_openai import ChatOpenAI model = ChatOpenAI() model.invoke('Say "Parrot-MAGIC"', stop=["-"]) # Should return `Parrot` # Using it the easy way via `bind` method which returns a new # RunnableBinding runnable_binding = model.bind(stop=["-"]) runnable_binding.invoke('Say "Parrot-MAGIC"') # Should return `Parrot` ``` Can also be done by instantiating a `RunnableBinding` directly (not recommended): ```python from langchain_core.runnables import RunnableBinding runnable_binding = RunnableBinding( bound=model, kwargs={"stop": ["-"]}, # <-- Note the additional kwargs ) runnable_binding.invoke('Say "Parrot-MAGIC"') # Should return `Parrot` ``` c URURURUR0UREUEUR UR S9$)zBind additional kwargs to a `Runnable`, returning a new `Runnable`. Args: **kwargs: The kwargs to bind to the `Runnable`. Returns: A new `Runnable` with the same type and config as the original, but with the additional kwargs bound. )rzrrrrr)rorzrrrrrr}s rxr~RunnableBinding.bindsT~~**;;!22,dkk,V,"44#66   r{Nc URURUR[S0UREU=(d 0EUE5UR UR URS9$)Nr)r>)rorzrrrrrrrs rxrRunnableBinding.with_configsf~~**;;(*UT[[*UV\r*Uf*UV!22"44#66   r{rc ^^^SUUU4SjjnURURURURU/URQUR UR S9$)aCBind lifecycle listeners to a `Runnable`, returning a new `Runnable`. The `Run` object contains information about the run, including its `id`, `type`, `input`, `output`, `error`, `start_time`, `end_time`, and any tags or metadata added to the run. Args: on_start: Called before the `Runnable` starts running, with the `Run` object. on_end: Called after the `Runnable` finishes running, with the `Run` object. on_error: Called if the `Runnable` throws an error, with the `Run` object. Returns: A new `Runnable` with the listeners bound. c">S[UTTTS9/0$rrrs rxlistener_config_factory?RunnableBinding.with_listeners..listener_config_factorys)'%!)%!)   r{r>)rr)rr)rorzrrrrr)rtrrrrs ``` rxrRunnableBinding.with_listenerss^@  ~~**;;;;5N8M8MN"44#66   r{c URURURURURUbUO UR UbUS9$UR S9$)Nr>rrs rxrRunnableBinding.with_types0sm ~~**;;;;!22(4 $:P:P +6   =AT"US[TRURSS550UD6$)Nr)r1rr0)r}rattrrts rxwrapper,RunnableBinding.__getattr__..wrapperUs;,T[[&**Xt:TU!r{c >[U5TS-:a/[U5n[TRUT5UT'T"U0UD6$T"US[TRUR SS550UD6$)Nrr)rrr1rr0)r}rargslrrrts rxrras|4yC!G+ $T %24;;c %Kc #U5f55,T[[&**Xt:TU!r{r)ryrzrrrzr{rkindr| KEYWORD_ONLYrPOSITIONAL_OR_KEYWORDrindex)rtrh config_paramrrrs` @@rx __getattr__RunnableBinding.__getattr__Lstzz4( D>>#--d3>>BB8L LL L  G$5$5$B$BBt  G$5$5$K$KK7,,T2==>DDXNt    r{rr@r rBrCrG)rryrrzrrA)rhrrr) rprrIrJrKr"r~rrrrrrOrr{rxr|r|s1f  ()-  %    !     2 2  2 2  !2 2 h6:7; 2 5  !   $  ""r{r|c\rSrSrSSjrSrg)_RunnableCallableSyncircgr rrt_inrs rx__call___RunnableCallableSync.__call__ssCr{rN)rr9rr)rr:rprrIrJrrOrr{rxrrrsOr{rc*\rSrSrSSjrSrg)_RunnableCallableAsyncivcgr rrs rxr_RunnableCallableAsync.__call__wsr{rN)rr9rr)rzAwaitable[Output]rrr{rxrrvs  (6  r{rc*\rSrSrSSjrSrg)_RunnableCallableIteratori|cgr rrs rxr"_RunnableCallableIterator.__call__}sr{rN)rr=rr)rr8rrr{rxrr|s"2@ r{rc*\rSrSrSSjrSrg)_RunnableCallableAsyncIteratoricgr rrs rxr'_RunnableCallableAsyncIterator.__call__s #r{rN)rr?rr)rr:rrr{rxrrs$'$7E$ $r{rcl[U[5(aU$[U5(d[R"U5(a [ U5$[ U5(a[[SU55$[U[5(a[S[U55$S[U53n[U5e)zCoerce a `Runnable`-like object into a `Runnable`. Args: thing: A `Runnable`-like object. Returns: A `Runnable`. Raises: TypeError: If the object is not `Runnable`-like. rrAzGExpected a Runnable, callable or dict.Instead got an unsupported type: ) rrerFrrrrprrrrrr^r)thingrs rxrrs%"" %  G$?$?$F$F ''d#>FGG%-/?/FGG ,,0K= : C.r{cgr rrs rxchainr"r{cgr rrs rxrrrr{cgr rrs rxrrrr{cgr rrs rxrrrr{c[U5$)a|Decorate a function to make it a `Runnable`. Sets the name of the `Runnable` to the name of the function. Any runnables called by the function will be traced as dependencies. Args: func: A `Callable`. Returns: A `Runnable`. Example: ```python from langchain_core.runnables import chain from langchain_core.prompts import PromptTemplate from langchain_openai import OpenAI @chain def my_func(fields): prompt = PromptTemplate("Hello, {name}!") model = OpenAI() formatted = prompt.invoke(**fields) for chunk in model.stream(formatted): yield chunk ``` )rrs rxrrsD $ r{)rvrrrrr)rr rrA)rz.Callable[[Input], Coroutine[Any, Any, Output]]rrA)rrrrA)rz(Callable[[Input], AsyncIterator[Output]]rrA)rrrrA)rzCallable[[Input], Output] | Callable[[Input], Iterator[Output]] | Callable[[Input], Coroutine[Any, Any, Output]] | Callable[[Input], AsyncIterator[Output]]rrA)rK __future__rrIrrVrrrQrrrcollections.abcrrrr r r r r concurrent.futuresrrr itertoolsroperatorrtypesrtypingrrrrrrrrrrpydanticrrr r!typing_extensionsr"langchain_core._apir# langchain_core.callbacks.managerr$r% langchain_core.load.serializabler&r'r(langchain_core.runnables.configr)r*r+r,r-r.r/r0r1r2r3r4langchain_core.runnables.utilsr5r6r7r8r9r:r;r<r=r>r?r@rArBrCrDrErF!langchain_core.tracers._streamingrG#langchain_core.tracers.event_streamrHrI!langchain_core.tracers.log_streamrKrL%langchain_core.tracers.root_listenersrMrNlangchain_core.utils.aiterrOrPlangchain_core.utils.iterrQlangchain_core.utils.pydanticrRrSrTrrVrrXRunnableWithFallbacksTrrZrr[langchain_core.runnables.schemar\rr]r^r_r`langchain_core.tracers.schemasrarbrrerQrsr~rrrrr RunnableMaprprrr rr<r|rrrrr rrrr{rxrs,1"#   5   =<&.R    (H6-9?5F;-EC2 Y& sGE6M*Y& xLL <%-)@L ^* #*-B**<0* #0*-B0*0*f !i +E6M:i XL+E4S>,ABL` l%/l%^ b XeVm,b JJ+DKf,EFJZr #E6M2r jO.uf}=Od M)%-8M`PHUF]3P XeVm4  7 $XeVm%<$ UF]w w &))*+ (6"2234 e$%}V'<<= > E6M* + UF]+ , v ./%UF]34c3h  6 " 8"" "  " -"" "  " 2"" "  " #"" " "  /"  " r{