langchain[patch],core[patch]: Fix agent executor stream event behavior

langchain-core/src/runnables/base.ts

@@ -823,7 +823,7 @@ export abstract class Runnable<
   ): AsyncGenerator<StreamEvent> {
     const eventStreamer = new EventStreamCallbackHandler({
       ...streamOptions,
-      autoClose: true,
+      autoClose: false,
     });
     const config = ensureConfig(options);
     const runId = config.runId ?? uuidv4();
@@ -843,14 +843,18 @@ export abstract class Runnable<
     // add each chunk to the output stream
     const outerThis = this;
     async function consumeRunnableStream() {
-      const runnableStream = await outerThis.stream(input, config);
-      const tappedStream = eventStreamer.tapOutputIterable(
-        runId,
-        runnableStream
-      );
-      // eslint-disable-next-line @typescript-eslint/no-unused-vars
-      for await (const _ of tappedStream) {
-        // Just iterate so that the callback handler picks up events
+      try {
+        const runnableStream = await outerThis.stream(input, config);
+        const tappedStream = eventStreamer.tapOutputIterable(
+          runId,
+          runnableStream
+        );
+        // eslint-disable-next-line @typescript-eslint/no-unused-vars
+        for await (const _ of tappedStream) {
+          // Just iterate so that the callback handler picks up events
+        }
+      } finally {
+        await eventStreamer.finish();
       }
     }
     const runnableStreamConsumePromise = consumeRunnableStream();

langchain-core/src/runnables/remote.ts

@@ -406,12 +406,13 @@ export class RemoteRunnable<
     const runManager = await callbackManager_?.handleChainStart(
       this.toJSON(),
       _coerceToDict(input, "input"),
+      config.runId,
       undefined,
       undefined,
       undefined,
-      undefined,
-      options?.runName
+      config.runName
     );
+    delete config.runId;
     let finalOutput: RunOutput | undefined;
     let finalOutputSupported = true;
     try {
@@ -476,12 +477,13 @@ export class RemoteRunnable<
     const runManager = await callbackManager_?.handleChainStart(
       this.toJSON(),
       _coerceToDict(input, "input"),
+      config.runId,
       undefined,
       undefined,
       undefined,
-      undefined,
-      options?.runName
+      config.runName
     );
+    delete config.runId;
     // The type is in camelCase but the API only accepts snake_case.
     const camelCaseStreamOptions = {
       include_names: streamOptions?.includeNames,
@@ -547,12 +549,13 @@ export class RemoteRunnable<
       const runManager = await callbackManager_?.handleChainStart(
         outerThis.toJSON(),
         _coerceToDict(input, "input"),
+        config.runId,
         undefined,
         undefined,
         undefined,
-        undefined,
-        options?.runName
+        config.runName
       );
+      delete config.runId;
       // The type is in camelCase but the API only accepts snake_case.
       const camelCaseStreamOptions = {
         include_names: streamOptions?.includeNames,

langchain-core/src/tracers/event_stream.ts

@@ -160,8 +160,6 @@ export class EventStreamCallbackHandler extends BaseTracer {
 
   protected excludeTags?: string[];
 
-  protected rootId?: string;
-
   private runInfoMap: Map<string, RunInfo> = new Map();
 
   private tappedPromises: Map<string, Promise<void>> = new Map();
@@ -238,7 +236,10 @@ export class EventStreamCallbackHandler extends BaseTracer {
       return;
     }
     const runInfo = this.runInfoMap.get(runId);
-    // run has finished, don't issue any stream events
+    // Run has finished, don't issue any stream events.
+    // An example of this is for runnables that use the default
+    // implementation of .stream(), which delegates to .invoke()
+    // and calls .onChainEnd() before passing it to the iterator.
     if (runInfo === undefined) {
       yield firstChunk.value;
       return;
@@ -286,7 +287,7 @@ export class EventStreamCallbackHandler extends BaseTracer {
       } finally {
         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
         tappedPromiseResolver!();
-        // Don't delete from the map to keep track of which runs have been tapped.
+        // Don't delete from the promises map to keep track of which runs have been tapped.
       }
     } else {
       // otherwise just pass through
@@ -607,18 +608,10 @@ export class EventStreamCallbackHandler extends BaseTracer {
     );
   }
 
-  async onRunCreate(run: Run): Promise<void> {
-    if (this.rootId === undefined) {
-      this.rootId = run.id;
-    }
-  }
-
-  async onRunUpdate(run: Run): Promise<void> {
-    if (run.id === this.rootId && this.autoClose) {
-      const pendingPromises = [...this.tappedPromises.values()];
-      void Promise.all(pendingPromises).finally(() => {
-        void this.writer.close();
-      });
-    }
+  async finish() {
+    const pendingPromises = [...this.tappedPromises.values()];
+    void Promise.all(pendingPromises).finally(() => {
+      void this.writer.close();
+    });
   }
 }

langchain/src/agents/executor.ts

@@ -174,12 +174,15 @@ export class AgentExecutorIterator
       this.runManager = await callbackManager?.handleChainStart(
         this.agentExecutor.toJSON(),
         this.inputs,
-        undefined,
+        this.config?.runId,
         undefined,
         this.tags ?? this.config?.tags,
         this.metadata ?? this.config?.metadata,
         this.runName ?? this.config?.runName
       );
+      if (this.config !== undefined) {
+        delete this.config.runId;
+      }
     }
   }
 
@@ -234,9 +237,7 @@ export class AgentExecutorIterator
           this.intermediateSteps,
           runManager
         );
-        if (this.runManager) {
-          await this.runManager.handleChainEnd(output);
-        }
+        await this.runManager?.handleChainEnd(output);
         await this.setFinalOutputs(output);
       }
     }
@@ -256,6 +257,7 @@ export class AgentExecutorIterator
       this.runManager
     );
     await this.setFinalOutputs(returnedOutput);
+    await this.runManager?.handleChainEnd(returnedOutput);
     return returnedOutput;
   }
 

langchain/src/agents/tests/create_tool_calling_agent.int.test.ts

@@ -1,9 +1,28 @@
+import { z } from "zod";
 import { test, expect } from "@jest/globals";
 import { ChatOpenAI } from "@langchain/openai";
 import { ChatPromptTemplate } from "@langchain/core/prompts";
+import { DynamicStructuredTool } from "@langchain/core/tools";
 import { TavilySearchResults } from "../../util/testing/tools/tavily_search.js";
 import { AgentExecutor, createToolCallingAgent } from "../index.js";
 
+const syntaxErrorTool = new DynamicStructuredTool({
+  name: "query",
+  description:
+    "use this tool to generate and execute a query from a question using the index.",
+  schema: z.object({
+    index_name: z.string().describe("The name of the index to query."),
+    question: z.string().describe("The question to answer."),
+  }),
+  func: async (_params) => {
+    return JSON.stringify({
+      result: "-ERR Syntax error at offset 19 near Bronx",
+      query:
+        'FT.AGGREGATE bites "@Borough:{The Bronx} @Gender:{M}" GROUPBY 0 REDUCE COUNT 0',
+    });
+  },
+});
+
 const tools = [new TavilySearchResults({ maxResults: 1 })];
 
 test("createToolCallingAgent works", async () => {
@@ -78,3 +97,44 @@ test("createToolCallingAgent stream events works", async () => {
     }
   }
 });
+
+test("createToolCallingAgent stream events works for multiple turns", async () => {
+  const prompt = ChatPromptTemplate.fromMessages([
+    ["system", "You are a helpful assistant"],
+    ["placeholder", "{chat_history}"],
+    ["human", "{input}"],
+    ["placeholder", "{agent_scratchpad}"],
+  ]);
+  const llm = new ChatOpenAI({
+    modelName: "gpt-4o",
+    temperature: 0,
+  });
+  const agent = await createToolCallingAgent({
+    llm,
+    tools: [syntaxErrorTool],
+    prompt,
+  });
+  const agentExecutor = new AgentExecutor({
+    agent,
+    tools: [syntaxErrorTool],
+    maxIterations: 3,
+  });
+  const input =
+    "Generate a query that looks up how many animals have been bitten in the Bronx.";
+  const eventStream = agentExecutor.streamEvents(
+    {
+      input,
+    },
+    {
+      version: "v2",
+    }
+  );
+
+  for await (const event of eventStream) {
+    const eventType = event.event;
+    console.log("Event type: ", eventType);
+    if (eventType === "on_chat_model_stream") {
+      console.log("Content: ", event.data);
+    }
+  }
+});

langchain/src/document_loaders/web/s3.ts

@@ -169,6 +169,7 @@ export class S3Loader extends BaseDocumentLoader {
       const docs = await unstructuredLoader.load();
 
       return docs;
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
     } catch (e: any) {
       throw new Error(
         `Failed to load file ${filePath} using unstructured loader: ${e.message}`

langchain/src/smith/config.ts

@@ -31,8 +31,11 @@ export type EvaluatorInputFormatter = ({
 }) => EvaluatorInputs;
 
 export type DynamicRunEvaluatorParams<
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
   Input extends Record<string, any> = Record<string, unknown>,
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
   Prediction extends Record<string, any> = Record<string, unknown>,
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
   Reference extends Record<string, any> = Record<string, unknown>
 > = {
   input: Input;