Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(algo): only return channels which actually triggered the task #324

Merged
merged 1 commit into from
Aug 16, 2024
Merged

fix(algo): only return channels which actually triggered the task #324

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions langgraph/src/pregel/algo.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,7 +337,7 @@ export function _prepareNextTasks<
// Check if any processes should be run in next step
// If so, prepare the values to be passed to them
for (const [name, proc] of Object.entries<PregelNode>(processes)) {
const hasUpdatedChannels = proc.triggers
const updatedChannels = proc.triggers
.filter((chan) => {
try {
readChannel(channels, chan, false);
Expand All @@ -346,11 +346,13 @@ export function _prepareNextTasks<
return false;
}
})
.some(
.filter(
(chan) =>
getChannelVersion(newCheckpoint, chan) >
getVersionSeen(newCheckpoint, name, chan)
);

const hasUpdatedChannels = updatedChannels.length > 0;
// If any of the channels read by this process were updated
if (hasUpdatedChannels) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
Expand Down Expand Up @@ -421,7 +423,7 @@ export function _prepareNextTasks<
const metadata = {
langgraph_step: extra.step,
langgraph_node: name,
langgraph_triggers: proc.triggers,
langgraph_triggers: updatedChannels,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ha good fix

langgraph_task_idx: tasks.length,
};
const checkpointNamespace =
Expand All @@ -438,7 +440,7 @@ export function _prepareNextTasks<
input: val,
proc: node,
writes,
triggers: proc.triggers,
triggers: updatedChannels,
config: patchConfig(mergeConfigs(proc.config, { metadata }), {
runName: name,
configurable: {
Expand Down
10 changes: 5 additions & 5 deletions langgraph/src/tests/pregel.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -757,15 +757,15 @@ describe("_prepareNextTasks", () => {
input: 100,
proc: new RunnablePassthrough(),
writes: [],
triggers: ["channel1", "channel2"],
triggers: ["channel1"],
config: {
tags: [],
configurable: expect.any(Object),
metadata: {
langgraph_node: "node2",
langgraph_step: -1,
langgraph_task_idx: 2,
langgraph_triggers: ["channel1", "channel2"],
langgraph_triggers: ["channel1"],
},
recursionLimit: 25,
runId: undefined,
Expand Down Expand Up @@ -2922,18 +2922,18 @@ it("checkpoint events", async () => {
timestamp: expect.any(String),
step: 3,
payload: {
id: "9b41e438-8b1c-5f01-b1cb-cc158ed8bd57",
id: "d082881c-b51f-5f07-a0f3-3f4581048aeb",
name: "finish",
input: { my_key: "value prepared slow", market: "DE" },
triggers: ["tool_two_fast", "tool_two_slow"],
triggers: ["tool_two_slow"],
},
},
{
type: "task_result",
timestamp: expect.any(String),
step: 3,
payload: {
id: "9b41e438-8b1c-5f01-b1cb-cc158ed8bd57",
id: "d082881c-b51f-5f07-a0f3-3f4581048aeb",
name: "finish",
result: [["my_key", " finished"]],
},
Expand Down
Loading