WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-23 08:02:23 +02:00
Code Issues Packages Projects Releases Wiki Activity

Lint

Browse Source
This commit is contained in:
Nuno Campos
2025-04-11 09:20:10 -07:00
parent 8ff5c43cf0
commit 5a7edead8c
1 changed files with 25 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/graph/state.py
+25 -26
View File
@@ -733,28 +733,6 @@ class CompiledStateGraph(CompiledGraph):
if is_writable_managed_value(v)
]
def _get_root(input: Any) -> Optional[Sequence[tuple[str, Any]]]:
if isinstance(input, Command):
if input.graph == Command.PARENT:
return ()
return input._update_as_tuples()
elif (
isinstance(input, (list, tuple))
and input
and any(isinstance(i, Command) for i in input)
):
updates: list[tuple[str, Any]] = []
for i in input:
if isinstance(i, Command):
if i.graph == Command.PARENT:
continue
updates.extend(i._update_as_tuples())
else:
updates.append(("__root__", i))
return updates
elif input is not None:
return [("__root__", input)]
def _get_updates(
input: Union[None, dict, Any],
) -> Optional[Sequence[tuple[str, Any]]]:
@@ -1083,7 +1061,7 @@ def _coerce_state(schema: Type[Any], input: dict[str, Any]) -> dict[str, Any]:
return schema(**input)
def _control_branch(value: Any, config: RunnableConfig) -> Sequence[Union[str, Send]]:
def _control_branch(value: Any, config: RunnableConfig) -> Any:
if isinstance(value, Send):
ChannelWrite.do_write(config, (value,))
return value
@@ -1114,9 +1092,7 @@ def _control_branch(value: Any, config: RunnableConfig) -> Sequence[Union[str, S
return value
async def _acontrol_branch(
value: Any, config: RunnableConfig
) -> Sequence[Union[str, Send]]:
async def _acontrol_branch(value: Any, config: RunnableConfig) -> Any:
if isinstance(value, Send):
ChannelWrite.do_write(config, (value,))
return value
@@ -1158,6 +1134,29 @@ CONTROL_BRANCH_PATH = RunnableCallable(
)
def _get_root(input: Any) -> Optional[Sequence[tuple[str, Any]]]:
if isinstance(input, Command):
if input.graph == Command.PARENT:
return ()
return input._update_as_tuples()
elif (
isinstance(input, (list, tuple))
and input
and any(isinstance(i, Command) for i in input)
):
updates: list[tuple[str, Any]] = []
for i in input:
if isinstance(i, Command):
if i.graph == Command.PARENT:
continue
updates.extend(i._update_as_tuples())
else:
updates.append(("__root__", i))
return updates
elif input is not None:
return [("__root__", input)]
def _get_channels(
schema: Type[dict],
) -> tuple[dict[str, BaseChannel], dict[str, ManagedValueSpec], dict[str, Any]]: