dispatchCustomEvent() doesn't work properly in tool.

[leSullivan]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

private async getDocuments(
        {
          query,
          reasoningStep,
        }: { query?: string; filter?: string; reasoningStep?: string },
        config: RunnableConfig,
      ) {
        if (!query) {
          throw new Error("Please provide the required 'query' property.");
        }
    
        this.streamReasoningStep(reasoningStep);
        .... 
    }

    async streamReasoningStep(reasoningStep: string) {
        if (!reasoningStep) {
          this.logger.warn('No reasoning step provided from', this.name);
          return;
        }
    
        await dispatchCustomEvent('onNewReasoningStep', {
          reasoningStep,
          agent: this.name,
        });
      }

Description

Hi im using a function Inside different Nodes of my Langgraph app to trigger a custom Event with dispatchCustomEvent().

I call this function after every tool call, but run into the error:

Error: Unable to dispatch a custom event without a parent run id. This function can only be called from within an existing run (e.g., inside a tool or a RunnableLambda).

Can someone help me with this problem ?

System Info

"@langchain/anthropic": "^0.2.15",
"@langchain/community": "^0.2.31",
"@langchain/core": "^0.2.28",
"@langchain/langgraph": "^0.1.2",
"@langchain/openai": "^0.2.7",

MacOS

Node 20.3.1

[dosubot[bot]]

Hey @leSullivan! 👋 I'm here to help you with your code, answer your questions, and provide support while waiting for a human maintainer. Let's take a look at the issue together.

The error occurs because dispatchCustomEvent() requires a parent run ID to be present in the callback manager. This function can only be called from within an existing run, such as inside a tool or a RunnableLambda.

To resolve this issue, ensure that you are calling dispatchCustomEvent() within the context of an existing run and that the config parameter is correctly passed. Here is an example of how to correctly use dispatchCustomEvent():

import { dispatchCustomEvent } from "@langchain/core/callbacks/dispatch";

const foo = RunnableLambda.from(async (input: string, config?: RunnableConfig) => {
  await dispatchCustomEvent(
    "my_custom_event",
    { arbitraryField: "someval" },
    config
  );
  return input;
});

const callbacks = [{
  handleCustomEvent: (eventName: string, payload: any) => {
    // Logs "my_custom_event" and { arbitraryField: "someval" }
    console.log(eventName, payload);
  }
}];

await foo.invoke("hi", { callbacks });

Ensure that the config object is correctly passed and that it contains the necessary context for the callback manager to retrieve the parent run ID. Additionally, make sure that the run context is properly managed and that the run_id is correctly assigned and propagated through the event stream ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[leSullivan]

@jacoblee93 do you have an Idea how to solve this problem ?

[jacoblee93]

It doesn't appear you're calling this in a tool?

Can you share the full context?

[leSullivan]

@jacoblee93
Sure I have a Langgraph chain with mutiple nodes. One of these nodes is a RAG Node with two tools.

 const chatChain = this.langgraphService.compileGraph();
 const stream = chatChain.streamEvents(initialState, {
   ...runnableConfig,
   version: 'v2',
 });


private async retrievelAgentNode(state: GraphState, config?: RunnableConfig) {
  this.logger.log('Retrieval Agent Thinking...');
  const agent = await this.retrievalAgent.compile();

  const lastMessage = state.messages[state.messages.length - 1];

const res: RetrievalAgentState = await agent.invoke(
  { messages: lastMessage },
  config,
);

const responseMessage: AIMessage = res.messages[res.messages.length - 1];
responseMessage.name = AgentName.RETRIEVAL;

this.logger.log('Retrieval Agent response:', responseMessage.content);

return { messages: [responseMessage], references: res.references };
}

  private getDocumentsTool = tool(this.getDocuments.bind(this), {
    name: NodeNames.GET_DOCUMENTS,
    description: 'Retrieve documents from the vectorstore.',
    schema: z.object({
      query: z
        .string()
        .describe(
          'A query that will be used to retrieve documents relevant for the task.',
        ),
      reasoningStep,
    }),
  });

So the example workflow is Input Message -> Graph -> Retrieval Agent -> getDocumentsNode() -> get DocumentsTool
Then the streamReasoningStep calls the dispatchCustomEvent. Every Invoke, Stream or StreamEvents is passed with my runnableConfig i define here

async initRunnableConfig(
    chatRequest: ChatRequestDto,
    auth: AuthInfo,
  ): Promise<RunnableConfig> {
    const langfuseHandler ...

    const llm = ...

    const configurable: ChatConfig = {
      llm,
      connections: ...,
    };

    return {
      configurable,
      callbacks: [langfuseHandler],
    };
  }

When i inspect the first Graph Action the runId already is undefined

[jacoblee93]

Ah is this in Nest.js? Someone reported an issue with using async hooks on Nest class methods that I still need to dig into.