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

Adjust stream output of Graph

Browse Source 
Now output of stream() are dicts where keys are node names and values are the output of that node on that step
This commit is contained in:
Nuno Campos
2024-01-08 14:48:26 -08:00
parent 142ad35965
commit 8c65b30ca7
7 changed files with 255 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files
langgraph/checkpoint/base.py
-7
View File
@@ -5,7 +5,6 @@ from datetime import datetime, timezone
from typing import Any, Optional, TypedDict
from langchain_core.load.serializable import Serializable
from langchain_core.pydantic_v1 import Field
from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.utils import ConfigurableFieldSpec
@@ -57,9 +56,3 @@ class BaseCheckpointSaver(Serializable, ABC):
return await asyncio.get_running_loop().run_in_executor(
None, self.put, config, checkpoint
)
class CheckpointView(Serializable):
values: dict[str, Any] = Field(frozen=True)
step: int
langgraph/graph/__init__.py
+21 -14
View File
@@ -11,6 +11,8 @@ from langchain_core.runnables.base import (
from langgraph.pregel import Channel, Pregel
END = "__end__"
class Branch(NamedTuple):
condition: Callable[..., str]
@@ -18,10 +20,8 @@ class Branch(NamedTuple):
def runnable(self, input: Any) -> Runnable:
result = self.condition(input)
return Channel.write_to(self.ends[result])
END = "__end__"
destination = self.ends[result]
return Channel.write_to(f"{destination}:inbox" if destination != END else END)
class Graph:
@@ -106,24 +106,31 @@ class Graph:
outgoing_edges = defaultdict(list)
for start, end in self.edges:
outgoing_edges[start].append(end)
outgoing_edges[start].append(f"{end}:inbox")
if hasattr(self, "finish_point"):
outgoing_edges[self.finish_point].append(END)
nodes = {
key: Channel.subscribe_to(key) | node for key, node in self.nodes.items()
key: (Channel.subscribe_to(f"{key}:inbox") | node | Channel.write_to(key))
for key, node in self.nodes.items()
}
for key, edges in outgoing_edges.items():
if edges:
nodes[key] |= Channel.write_to(*edges)
for key, branches in self.branches.items():
for branch in branches:
nodes[key] |= RunnableLambda(branch.runnable, name=f"{key}_condition")
for key in self.nodes:
outgoing = outgoing_edges[key]
edges_key = f"{key}:edges"
if outgoing or key in self.branches:
nodes[edges_key] = Channel.subscribe_to(key)
if outgoing:
nodes[edges_key] |= Channel.write_to(*[dest for dest in outgoing])
if key in self.branches:
for branch in self.branches[key]:
nodes[edges_key] |= RunnableLambda(
branch.runnable, name=f"{key}_condition"
)
return Pregel(
nodes=nodes,
input=self.entry_point,
input=f"{self.entry_point}:inbox",
output=END,
hidden=[f"{node}:inbox" for node in self.nodes],
)
langgraph/pregel/__init__.py
+27 -71
View File
@@ -53,7 +53,6 @@ from langgraph.checkpoint.base import (
BaseCheckpointSaver,
Checkpoint,
CheckpointAt,
CheckpointView,
empty_checkpoint,
)
from langgraph.constants import CONFIG_KEY_READ, CONFIG_KEY_SEND
@@ -153,6 +152,8 @@ class Pregel(
output: Union[str, Sequence[str]] = "output"
hidden: Sequence[str] = Field(default_factory=list)
input: Union[str, Sequence[str]] = "input"
step_timeout: Optional[float] = None
@@ -220,11 +221,12 @@ class Pregel(
config: RunnableConfig,
*,
output: Optional[Union[str, Sequence[str]]] = None,
) -> Iterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
) -> Iterator[Union[dict[str, Any], Any]]:
if config["recursion_limit"] < 1:
raise ValueError("recursion_limit must be at least 1")
# assign defaults
output = output if output is not None else [chan for chan in self.channels]
if output is None:
output = [chan for chan in self.channels if chan not in self.hidden]
# copy nodes to ignore mutations during execution
processes = {**self.nodes}
# get checkpoint from saver, or create an empty one
@@ -303,13 +305,12 @@ class Pregel(
print_checkpoint(step, channels)
# yield current value and checkpoint view
view = CheckpointView(
values=_updateable_channel_values(channels),
step=step + 1,
)
yield map_output(output, pending_writes, channels), view
# if view was updated, apply writes to channels
_apply_writes_from_view(checkpoint, channels, view)
if step_output := map_output(output, pending_writes, channels):
yield step_output
# we can detect updates when output is multiple channels (ie. dict)
if not isinstance(output, str):
# if view was updated, apply writes to channels
_apply_writes_from_view(checkpoint, channels, step_output)
# save end of step checkpoint
if self.saver is not None and self.saver.at == CheckpointAt.END_OF_STEP:
@@ -328,7 +329,7 @@ class Pregel(
config: RunnableConfig,
*,
output: Optional[Union[str, Sequence[str]]] = None,
) -> AsyncIterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
) -> AsyncIterator[Union[dict[str, Any], Any]]:
if config["recursion_limit"] < 1:
raise ValueError("recursion_limit must be at least 1")
# if running from astream_log() run each proc with streaming
@@ -341,7 +342,8 @@ class Pregel(
None,
)
# assign defaults
output = output if output is not None else [chan for chan in self.channels]
if output is None:
output = [chan for chan in self.channels if chan not in self.hidden]
# copy nodes to ignore mutations during execution
processes = {**self.nodes}
# get checkpoint from saver, or create an empty one
@@ -423,13 +425,12 @@ class Pregel(
print_checkpoint(step, channels)
# yield current value and checkpoint view
view = CheckpointView(
values=_updateable_channel_values(channels),
step=step + 1,
)
yield map_output(output, pending_writes, channels), view
# if view was updated, apply writes to channels
_apply_writes_from_view(checkpoint, channels, view)
if step_output := map_output(output, pending_writes, channels):
yield step_output
# we can detect updates when output is multiple channels (ie. dict)
if not isinstance(output, str):
# if view was updated, apply writes to channels
_apply_writes_from_view(checkpoint, channels, step_output)
# save end of step checkpoint
if self.saver is not None and self.saver.at == CheckpointAt.END_OF_STEP:
@@ -477,24 +478,10 @@ class Pregel(
output: Optional[Union[str, Sequence[str]]] = None,
**kwargs: Any,
) -> Iterator[Union[dict[str, Any], Any]]:
for out, _ in self._transform_stream_with_config(
for chunk in self._transform_stream_with_config(
input, self._transform, config, output=output, **kwargs
):
if out is not None:
yield cast(Union[dict[str, Any], Any], out)
def step(
self,
input: Union[dict[str, Any], Any],
config: Optional[RunnableConfig] = None,
*,
output: Optional[Union[str, Sequence[str]]] = None,
**kwargs: Any,
) -> Iterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
for tup in self._transform_stream_with_config(
iter([input]), self._transform, config, output=output, **kwargs
):
yield cast(tuple[Union[dict[str, Any], Any], CheckpointView], tup)
yield chunk
async def ainvoke(
self,
@@ -538,27 +525,10 @@ class Pregel(
output: Optional[Union[str, Sequence[str]]] = None,
**kwargs: Any,
) -> AsyncIterator[Union[dict[str, Any], Any]]:
async for out, _ in self._atransform_stream_with_config(
async for chunk in self._atransform_stream_with_config(
input, self._atransform, config, output=output, **kwargs
):
if out is not None:
yield out
async def astep(
self,
input: Union[dict[str, Any], Any],
config: Optional[RunnableConfig] = None,
*,
output: Optional[Union[str, Sequence[str]]] = None,
**kwargs: Any,
) -> AsyncIterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
async def input_stream() -> AsyncIterator[Union[dict[str, Any], Any]]:
yield input
async for tup in self._atransform_stream_with_config(
input_stream(), self._atransform, config, output=output, **kwargs
):
yield cast(tuple[Union[dict[str, Any], Any], CheckpointView], tup)
yield chunk
def _interrupt_or_proceed(
@@ -629,11 +599,9 @@ def _apply_writes(
def _apply_writes_from_view(
checkpoint: Checkpoint,
channels: Mapping[str, BaseChannel],
view: CheckpointView,
checkpoint: Checkpoint, channels: Mapping[str, BaseChannel], values: dict[str, Any]
) -> None:
for chan, value in view.values.items():
for chan, value in values.items():
if value == channels[chan].get():
continue
@@ -642,7 +610,7 @@ def _apply_writes_from_view(
f"{channels[chan].__class__.__name__}"
)
checkpoint["channel_versions"][chan] += 1
channels[chan].update([view.values[chan]])
channels[chan].update([values[chan]])
def _prepare_next_tasks(
@@ -703,18 +671,6 @@ def _prepare_next_tasks(
return tasks
def _updateable_channel_values(channels: Mapping[str, BaseChannel]) -> dict[str, Any]:
"""Return a dictionary of updateable channel values."""
values: dict[str, Any] = {}
for k, v in channels.items():
if isinstance(v, LastValue) and k not in [c.value for c in ReservedChannels]:
try:
values[k] = v.get()
except EmptyChannelError:
pass
return values
async def _aconsume(iterator: AsyncIterator[Any]) -> None:
"""Consume an async iterator."""
async for _ in iterator:
langgraph/pregel/write.py
+3
View File
@@ -32,6 +32,9 @@ class ChannelWrite(RunnablePassthrough):
super().__init__(func=self._write, afunc=self._awrite, channels=channels)
self.name = f"ChannelWrite<{','.join(chan for chan, _ in self.channels)}>"
def __repr_args__(self) -> Any:
return [("channels", self.channels)]
@property
def config_specs(self) -> list[ConfigurableFieldSpec]:
return [
poetry.lock
Generated
+3 -53
View File
@@ -1,4 +1,4 @@
# This file is automatically @generated by Poetry 1.6.1 and should not be changed by hand.
# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand.
[[package]]
name = "aiohttp"
@@ -121,9 +121,6 @@ files = [
{file = "annotated_types-0.6.0.tar.gz", hash = "sha256:563339e807e53ffd9c267e99fc6d9ea23eb8443c08f112651963e24e22f84a5d"},
]
[package.dependencies]
typing-extensions = {version = ">=4.0.0", markers = "python_version < \"3.9\""}
[[package]]
name = "anyio"
version = "4.2.0"
@@ -306,9 +303,6 @@ files = [
{file = "Babel-2.14.0.tar.gz", hash = "sha256:6919867db036398ba21eb5c7a0f6b28ab8cbc3ae7a73a44ebe34ae74a4e7d363"},
]
[package.dependencies]
pytz = {version = ">=2015.7", markers = "python_version < \"3.9\""}
[package.extras]
dev = ["freezegun (>=1.0,<2.0)", "pytest (>=6.0)", "pytest-cov"]
@@ -988,24 +982,6 @@ docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.link
perf = ["ipython"]
testing = ["flufl.flake8", "importlib-resources (>=1.3)", "packaging", "pyfakefs", "pytest (>=6)", "pytest-black (>=0.3.7)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-mypy (>=0.9.1)", "pytest-perf (>=0.9.2)", "pytest-ruff"]
[[package]]
name = "importlib-resources"
version = "6.1.1"
description = "Read resources from Python packages"
optional = false
python-versions = ">=3.8"
files = [
{file = "importlib_resources-6.1.1-py3-none-any.whl", hash = "sha256:e8bf90d8213b486f428c9c39714b920041cb02c184686a3dee24905aaa8105d6"},
{file = "importlib_resources-6.1.1.tar.gz", hash = "sha256:3893a00122eafde6894c59914446a512f728a0c1a45f9bb9b63721b6bacf0b4a"},
]
[package.dependencies]
zipp = {version = ">=3.1.0", markers = "python_version < \"3.10\""}
[package.extras]
docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.linker (>=1.9)", "sphinx (<7.2.5)", "sphinx (>=3.5)", "sphinx-lint"]
testing = ["pytest (>=6)", "pytest-black (>=0.3.7)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-mypy (>=0.9.1)", "pytest-ruff", "zipp (>=3.17)"]
[[package]]
name = "iniconfig"
version = "2.0.0"
@@ -1214,11 +1190,9 @@ files = [
attrs = ">=22.2.0"
fqdn = {version = "*", optional = true, markers = "extra == \"format-nongpl\""}
idna = {version = "*", optional = true, markers = "extra == \"format-nongpl\""}
importlib-resources = {version = ">=1.4.0", markers = "python_version < \"3.9\""}
isoduration = {version = "*", optional = true, markers = "extra == \"format-nongpl\""}
jsonpointer = {version = ">1.13", optional = true, markers = "extra == \"format-nongpl\""}
jsonschema-specifications = ">=2023.03.6"
pkgutil-resolve-name = {version = ">=1.3.10", markers = "python_version < \"3.9\""}
referencing = ">=0.28.4"
rfc3339-validator = {version = "*", optional = true, markers = "extra == \"format-nongpl\""}
rfc3986-validator = {version = ">0.1.0", optional = true, markers = "extra == \"format-nongpl\""}
@@ -1242,7 +1216,6 @@ files = [
]
[package.dependencies]
importlib-resources = {version = ">=1.4.0", markers = "python_version < \"3.9\""}
referencing = ">=0.31.0"
[[package]]
@@ -1441,7 +1414,6 @@ files = [
[package.dependencies]
async-lru = ">=1.0.0"
importlib-metadata = {version = ">=4.8.3", markers = "python_version < \"3.10\""}
importlib-resources = {version = ">=1.4", markers = "python_version < \"3.9\""}
ipykernel = "*"
jinja2 = ">=3.0.3"
jupyter-core = "*"
@@ -2147,17 +2119,6 @@ files = [
{file = "pickleshare-0.7.5.tar.gz", hash = "sha256:87683d47965c1da65cdacaf31c8441d12b8044cdec9aca500cd78fc2c683afca"},
]
[[package]]
name = "pkgutil-resolve-name"
version = "1.3.10"
description = "Resolve a name to an object."
optional = false
python-versions = ">=3.6"
files = [
{file = "pkgutil_resolve_name-1.3.10-py3-none-any.whl", hash = "sha256:ca27cc078d25c5ad71a9de0a7a330146c4e014c2462d9af19c6b828280649c5e"},
{file = "pkgutil_resolve_name-1.3.10.tar.gz", hash = "sha256:357d6c9e6a755653cfd78893817c0853af365dd51ec97f3d358a819373bbd174"},
]
[[package]]
name = "platformdirs"
version = "4.1.0"
@@ -2575,17 +2536,6 @@ files = [
{file = "python_json_logger-2.0.7-py3-none-any.whl", hash = "sha256:f380b826a991ebbe3de4d897aeec42760035ac760345e57b812938dc8b35e2bd"},
]
[[package]]
name = "pytz"
version = "2023.3.post1"
description = "World timezone definitions, modern and historical"
optional = false
python-versions = "*"
files = [
{file = "pytz-2023.3.post1-py2.py3-none-any.whl", hash = "sha256:ce42d816b81b68506614c11e8937d3aa9e41007ceb50bfdcb0749b921bf646c7"},
{file = "pytz-2023.3.post1.tar.gz", hash = "sha256:7b4fddbeb94a1eba4b557da24f19fdf9db575192544270a9101d8509f9f43d7b"},
]
[[package]]
name = "pywin32"
version = "306"
@@ -3617,5 +3567,5 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
[metadata]
lock-version = "2.0"
python-versions = ">=3.8.1,<4.0"
content-hash = "9aa1c4dcb484322ed2c5fc403902227865a825c9f1fd8874563edff0432133b6"
python-versions = ">=3.9.0,<4.0"
content-hash = "859cb37bda0ed0d3e8f94df41f0f6c5785aaf144e4e1dbb3835b945aa958ef1e"
tests/test_pregel.py
+97 -57
View File
@@ -156,38 +156,28 @@ def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
assert app.invoke(2) == 4
for output, view in app.step(2):
if view.step == 1:
assert view.values == {
for step, values in enumerate(app.stream(2), start=1):
if step == 1:
assert values == {
"inbox": 3,
"input": 2,
}
assert output == {"inbox": 3}
elif view.step == 2:
assert view.values == {
elif step == 2:
assert values == {
"output": 4,
"inbox": 3,
"input": 2,
}
assert output == {"output": 4}
for output, view in app.step(2):
if view.step == 1:
assert view.values == {
for step, values in enumerate(app.stream(2), start=1):
if step == 1:
assert values == {
"inbox": 3,
"input": 2,
}
assert output == {"inbox": 3}
# modify inbox value
view.values["inbox"] = 5
elif view.step == 2:
assert view.values == {
"output": 6,
"inbox": 5,
"input": 2,
}
values["inbox"] = 5
elif step == 2:
# output is different now
assert output == {"output": 6}
assert values == {
"output": 6,
}
graph = Graph()
graph.add_node("add_one", add_one)
@@ -199,38 +189,41 @@ def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
assert gapp.invoke(2) == 4
for output, view in gapp.step(2):
if view.step == 1:
assert view.values == {
"add_one": 2,
"add_one_more": 3,
for step, values in enumerate(gapp.stream(2), start=1):
if step == 1:
assert values == {
"add_one": 3,
}
assert output == {"add_one_more": 3}
elif view.step == 2:
assert view.values == {
"add_one": 2,
"add_one_more": 3,
elif step == 2:
assert values == {
"add_one_more": 4,
}
elif step == 3:
assert values == {
"__end__": 4,
}
assert output == {"__end__": 4}
else:
assert 0, f"{step}:{values}"
assert step == 3
for output, view in gapp.step(2):
if view.step == 1:
assert view.values == {
"add_one": 2,
"add_one_more": 3,
for step, values in enumerate(gapp.stream(2), start=1):
if step == 1:
assert values == {
"add_one": 3,
}
assert output == {"add_one_more": 3}
# modify inbox value
view.values["add_one_more"] = 5
elif view.step == 2:
assert view.values == {
"add_one": 2,
"add_one_more": 5,
# modify value before next step
values["add_one"] = 5
elif step == 2:
assert values == {
"add_one_more": 6,
}
elif step == 3:
assert values == {
"__end__": 6,
}
# output is different now
assert output == {"__end__": 6}
else:
assert 0, "Should not get here"
assert step == 3
def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None:
@@ -658,14 +651,9 @@ def test_conditional_graph() -> None:
),
}
assert [
deepcopy(c)
for c in app.stream(
{"input": "what is weather in sf"}, output=["agent", "tools"]
)
] == [
assert [deepcopy(c) for c in app.stream({"input": "what is weather in sf"})] == [
{
"tools": {
"agent": {
"input": "what is weather in sf",
"agent_outcome": AgentAction(
tool="search_api", tool_input="query", log="tool:search_api:query"
@@ -673,7 +661,7 @@ def test_conditional_graph() -> None:
}
},
{
"agent": {
"tools": {
"input": "what is weather in sf",
"intermediate_steps": [
(
@@ -688,7 +676,7 @@ def test_conditional_graph() -> None:
}
},
{
"tools": {
"agent": {
"input": "what is weather in sf",
"intermediate_steps": [
(
@@ -707,6 +695,29 @@ def test_conditional_graph() -> None:
),
}
},
{
"tools": {
"input": "what is weather in sf",
"intermediate_steps": [
(
AgentAction(
tool="search_api",
tool_input="query",
log="tool:search_api:query",
),
"result for query",
),
(
AgentAction(
tool="search_api",
tool_input="another",
log="tool:search_api:another",
),
"result for another",
),
],
}
},
{
"agent": {
"input": "what is weather in sf",
@@ -728,6 +739,35 @@ def test_conditional_graph() -> None:
"result for another",
),
],
"agent_outcome": AgentFinish(
return_values={"answer": "answer"}, log="finish:answer"
),
}
},
{
"__end__": {
"input": "what is weather in sf",
"intermediate_steps": [
(
AgentAction(
tool="search_api",
tool_input="query",
log="tool:search_api:query",
),
"result for query",
),
(
AgentAction(
tool="search_api",
tool_input="another",
log="tool:search_api:another",
),
"result for another",
),
],
"agent_outcome": AgentFinish(
return_values={"answer": "answer"}, log="finish:answer"
),
}
},
]
tests/test_pregel_async.py
+104 -55
View File
@@ -157,38 +157,34 @@ async def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
assert await app.ainvoke(2) == 4
async for output, view in app.astep(2):
if view.step == 1:
assert view.values == {
step = 0
async for values in app.astream(2):
step += 1
if step == 1:
assert values == {
"inbox": 3,
"input": 2,
}
assert output == {"inbox": 3}
elif view.step == 2:
assert view.values == {
elif step == 2:
assert values == {
"output": 4,
"inbox": 3,
"input": 2,
}
assert output == {"output": 4}
assert step == 2
async for output, view in app.astep(2):
if view.step == 1:
assert view.values == {
step = 0
async for values in app.astream(2):
step += 1
if step == 1:
assert values == {
"inbox": 3,
"input": 2,
}
assert output == {"inbox": 3}
# modify inbox value
view.values["inbox"] = 5
elif view.step == 2:
assert view.values == {
"output": 6,
"inbox": 5,
"input": 2,
}
values["inbox"] = 5
elif step == 2:
# output is different now
assert output == {"output": 6}
assert values == {
"output": 6,
}
assert step == 2
graph = Graph()
graph.add_node("add_one", add_one)
@@ -200,38 +196,42 @@ async def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
assert await gapp.ainvoke(2) == 4
async for output, view in gapp.astep(2):
if view.step == 1:
assert view.values == {
"add_one": 2,
"add_one_more": 3,
step = 0
async for values in gapp.astream(2):
step += 1
if step == 1:
assert values == {
"add_one": 3,
}
assert output == {"add_one_more": 3}
elif view.step == 2:
assert view.values == {
"add_one": 2,
"add_one_more": 3,
elif step == 2:
assert values == {
"add_one_more": 4,
}
elif step == 3:
assert values == {
"__end__": 4,
}
assert output == {"__end__": 4}
assert step == 3
async for output, view in gapp.astep(2):
if view.step == 1:
assert view.values == {
"add_one": 2,
"add_one_more": 3,
step = 0
async for values in gapp.astream(2):
step += 1
if step == 1:
assert values == {
"add_one": 3,
}
assert output == {"add_one_more": 3}
# modify inbox value
view.values["add_one_more"] = 5
elif view.step == 2:
assert view.values == {
"add_one": 2,
"add_one_more": 5,
# modify value before running next step
values["add_one"] = 5
elif step == 2:
# output is different now
assert values == {
"add_one_more": 6,
}
elif step == 3:
assert values == {
"__end__": 6,
}
# output is different now
assert output == {"__end__": 6}
assert step == 3
async def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None:
@@ -690,13 +690,10 @@ async def test_conditional_graph() -> None:
}
assert [
deepcopy(c)
async for c in app.astream(
{"input": "what is weather in sf"}, output=["agent", "tools"]
)
deepcopy(c) async for c in app.astream({"input": "what is weather in sf"})
] == [
{
"tools": {
"agent": {
"input": "what is weather in sf",
"agent_outcome": AgentAction(
tool="search_api", tool_input="query", log="tool:search_api:query"
@@ -704,7 +701,7 @@ async def test_conditional_graph() -> None:
}
},
{
"agent": {
"tools": {
"input": "what is weather in sf",
"intermediate_steps": [
(
@@ -719,7 +716,7 @@ async def test_conditional_graph() -> None:
}
},
{
"tools": {
"agent": {
"input": "what is weather in sf",
"intermediate_steps": [
(
@@ -738,6 +735,29 @@ async def test_conditional_graph() -> None:
),
}
},
{
"tools": {
"input": "what is weather in sf",
"intermediate_steps": [
(
AgentAction(
tool="search_api",
tool_input="query",
log="tool:search_api:query",
),
"result for query",
),
(
AgentAction(
tool="search_api",
tool_input="another",
log="tool:search_api:another",
),
"result for another",
),
],
}
},
{
"agent": {
"input": "what is weather in sf",
@@ -759,6 +779,35 @@ async def test_conditional_graph() -> None:
"result for another",
),
],
"agent_outcome": AgentFinish(
return_values={"answer": "answer"}, log="finish:answer"
),
}
},
{
"__end__": {
"input": "what is weather in sf",
"intermediate_steps": [
(
AgentAction(
tool="search_api",
tool_input="query",
log="tool:search_api:query",
),
"result for query",
),
(
AgentAction(
tool="search_api",
tool_input="another",
log="tool:search_api:another",
),
"result for another",
),
],
"agent_outcome": AgentFinish(
return_values={"answer": "answer"}, log="finish:answer"
),
}
},
]