community[patch]: anthropic add tool call support new tools api

[LordMsz]

This is a proposal to add support for latest tools call API for Anthropic models on Amazon Bedrock in LangChain / Community.
I originally developed my own solution, but then realized it basically matches solution in https://github.com/LordMsz/langchainjs/blob/main/libs/langchain-anthropic/src/chat_models.ts .
So I tried to sync as much as sensible as part of this change.

I'd like to ask for help getting this finalized, as I don't have experience contributing to this repo or LangChain in general.

I tested the code locally running following code:

import { config } from 'dotenv';
// import { BedrockChat } from "@langchain/community/chat_models/bedrock";
import { BedrockChat } from "./custom-models/index"; // this is my upadted version of BedrockChat that I want to propose as a PR
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { z } from "zod";
import { createToolCallingAgent, AgentExecutor } from "langchain/agents";
import { DynamicStructuredTool } from "@langchain/core/tools";

config(); // load .env file

if (!process?.env?.AWSAccessKeyId || !process?.env?.AWSSecretAccessKey) {
    console.error('AWSAccessKeyId or AWSSecretAccessKey not found in .env file');
    process.exit(1);
}

const addTool = new DynamicStructuredTool({
    name: "add",
    description: "Add two integers together.",
    schema: z.object({
        firstInt: z.number(),
        secondInt: z.number(),
    }),
    func: async ({ firstInt, secondInt }) => {
        console.log('------ Function called with: -----------', { firstInt, secondInt });
        return (firstInt + secondInt).toString();
    },
});

async function test() {
    const model = new BedrockChat({
        model: "anthropic.claude-3-sonnet-20240229-v1:0",
        temperature: 0.5,
        region: "us-east-1",
        credentials: {
            accessKeyId: process.env.AWSAccessKeyId as string,
            secretAccessKey: process.env.AWSSecretAccessKey as string,
        },
    });
    
    const prompt = ChatPromptTemplate.fromMessages([
        ["system", "You are a helpful assistant"],
        ["placeholder", "{chat_history}"],
        ["human", "{input}"],
        ["placeholder", "{agent_scratchpad}"],
    ])
    
    const agent = createToolCallingAgent({
        llm: model,
        tools: [addTool],
        prompt: prompt,
        streamRunnable: false // attempt to work around streaming issue with tools
    });
    
    const agentExecutor = new AgentExecutor({ 
        agent,
        tools: [addTool],
        // verbose: true,
        returnIntermediateSteps: true,
    });
    
    const r = await agentExecutor.invoke({ input: "Hi! Can you tell me what is 10 + 4?" });
    console.log(r);
    console.log(r.content);
}

test();

Fixes #5639 (issue)

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

Name	Status	Preview	Comments	Updated (UTC)
langchainjs-api-refs	✅ Ready (Inspect)	Visit Preview	💬 Add feedback	Jun 7, 2024 2:17am
langchainjs-docs	✅ Ready (Inspect)	Visit Preview		Jun 7, 2024 2:17am

[jakowenko]

Thanks for putting this together! Make sure you format and lint the code as a large portion of the diff seems to be related to that.

[LordMsz]

Apologies for the linting and formatting issue - fixed now.
I struggle to get the complete build and test of project working for some reason, but I tried all the changes with copy of the models. Any help with proper build highly appreciated!

[jacoblee93]

This is massive - thank you so much for this!

Will have a look tomorrow.

[jacoblee93]

I wonder if there's a way to reuse some of the existing logic we have in @langchain/anthropic - might have a short look to avoid duplication.

[LordMsz]

Hi,
sure, I was also thinking that way - was surprised that big chunks of code are mostly a verbatim copy from @langchain/anthropic.
Ideally we would re-use it all and in Bedrock wrapper only authentication would be handled?

What is your suggestion? Should I expose the utilized methods in @langchain/anthropic, e.g. _mergeMessages. and import them here, or something else?

[jacoblee93]

Yes that's what I was thinking - if you're willing to take it on I think that'd be a good approach!

CC @efriis to weigh in as well.

[LordMsz]

Okay, working on it now.
I'm also trying to resolve my local build issue - seems I was missing Python 3.11 (required for node-gyp) and now I need C++ Workload in VS to add "msvs_version".
Is this documented anywhere as a per-requisite for development? Sorry if I missed it.

[LordMsz]

This change is biggest question for me - if it's inline with overall "strategy" that I can export this from Anthropic and use it in community? Any thoughts?
I only need single function from there, so that could be a minor copy-paste I assume.

[jacoblee93]

Yeah I think that's fine. Maybe we can prefix it with an underscore to show that it shouldn't be imported directly.

[LordMsz]

Ok, I got into a bit complex territory.

 Invalid package dependency graph: cyclic dependency detected:
  │     @langchain/community, @langchain/anthropic
  ╰─▶ cyclic dependency detected:
        @langchain/community, @langchain/anthropic

So community is used in Anthropic but I need it the other way around now.
When I simply copy-over the code into my test project, everything works code-wise.
I've cleaned up some linting errors along the way.

Any suggestions please? Any help appreciated!

Btw: I finally made local build work somewhat through dev-container; couldn't make it run on Windows at all.

[LordMsz]

I would propose to copy the required functions with TODO to refactor them to some common place both for Anthropic and Community?
I don't know the repo well enough (yet? :)) to be able to decide this on my own.

[jacoblee93]

Thanks for looking into this! I can resolve the build issue, it should just be dev dependencies I think.

[jacoblee93]

Thank you for the awesome start!

Will try to land tonight

[kamilcglr]

Hello, thank you all for the quick integration.

Is it normal for streaming not to work when using the createToolCallingAgent function? I see that commit contains the term streaming, but I don't understand if it's just about disabling calling tools in an async way or all events.

My use case is as follows: I am using the stream function and I want to send all the generations of the final response token by token to my client using websocket using the 'on_llm_stream' event.

for await (const event of eventStream) { ...

[jacoblee93]

Hey @kamilcglr, are you talking about streamEvents?

Proper streaming tool calls will take a bit more work - the delegation you see in that commit just invokes and yields the non-streaming result from Bedrock as a single chunk. You should still see it as an on_chat_model_stream event though.

[kamilcglr]

Hello @jacoblee93

Yes, I am talking about streamEvents, but I do not necessarily want to stream the tools, just the final answer.

Here is an example of what I am trying to do. I have a class with a run function that starts the process and sends event to my frontend application.
I used to use a similar class in Python with an SQL Agent and Bedrock with Claude 3. I managed to send all the tokens to my frontend one by one (with the on_chat_model_stream event).

class AgentService {
  private agent

  constructor() {}

  async initialize() {
    if (this.agent) return

    const llm = new BedrockChat({
      model: 'anthropic.claude-3-sonnet-20240229-v1:0',
      region: 'us-west-2',
      streaming: true,
      temperature: 0,
    })

    const tools = [
      ... some DynamicStructuredTools ...
    ]

    const agent = createToolCallingAgent({
      llm,
      tools,
      prompt: customPrompt,
      streamRunnable: true,
    })

    const agentExecutor = new AgentExecutor({
      agent,
      tools,
      handleParsingErrors: true,
      returnIntermediateSteps: true,
    }).withConfig({ runName: 'llm_agent' })

    const messageHistory = (id: string) =>
      new DynamoDBChatMessageHistory({
        tableName: 'SessionTable',
        partitionKey: 'SessionId',
        sessionId: id,
        config: {
          region: 'us-west-2',
        },
      })

    const runnableWithMessageHistory = new RunnableWithMessageHistory({
      runnable: agentExecutor,
      getMessageHistory: (sessionId: string) => messageHistory(sessionId),
      inputMessagesKey: 'input',
      historyMessagesKey: 'history',
    })

    this.agent = runnableWithMessageHistory
  }

  async run(input: string, sessionId: string) {
    // [This is where an event is sent to the client to say that the process has started]

    const eventStream = await this.agent.streamEvents(
      { input: input },
      {
        configurable: {
          sessionId: sessionId,
        },
        version: 'v1',
      }
    )

    for await (const item of eventStream) {
      if (item.event === 'on_llm_stream') {
        // Here I send "item.data.chunk.content" to the client as a completion
        // But it sends everything in a single event, whereas I expect something like: Hello,| how |can |I |help |you ? where ‘|’ is an event
      }

      if (item.event === 'on_chain_end') {
       // [End event is sent]
      }
    }
  }
}

[Moe03]

Hello @jacoblee93

Yes, I am talking about streamEvents, but I do not necessarily want to stream the tools, just the final answer.

Here is an example of what I am trying to do. I have a class with a run function that starts the process and sends event to my frontend application. I used to use a similar class in Python with an SQL Agent and Bedrock with Claude 3. I managed to send all the tokens to my frontend one by one (with the on_chat_model_stream event).

class AgentService {
  private agent

  constructor() {}

  async initialize() {
    if (this.agent) return

    const llm = new BedrockChat({
      model: 'anthropic.claude-3-sonnet-20240229-v1:0',
      region: 'us-west-2',
      streaming: true,
      temperature: 0,
    })

    const tools = [
      ... some DynamicStructuredTools ...
    ]

    const agent = createToolCallingAgent({
      llm,
      tools,
      prompt: customPrompt,
      streamRunnable: true,
    })

    const agentExecutor = new AgentExecutor({
      agent,
      tools,
      handleParsingErrors: true,
      returnIntermediateSteps: true,
    }).withConfig({ runName: 'llm_agent' })

    const messageHistory = (id: string) =>
      new DynamoDBChatMessageHistory({
        tableName: 'SessionTable',
        partitionKey: 'SessionId',
        sessionId: id,
        config: {
          region: 'us-west-2',
        },
      })

    const runnableWithMessageHistory = new RunnableWithMessageHistory({
      runnable: agentExecutor,
      getMessageHistory: (sessionId: string) => messageHistory(sessionId),
      inputMessagesKey: 'input',
      historyMessagesKey: 'history',
    })

    this.agent = runnableWithMessageHistory
  }

  async run(input: string, sessionId: string) {
    // [This is where an event is sent to the client to say that the process has started]

    const eventStream = await this.agent.streamEvents(
      { input: input },
      {
        configurable: {
          sessionId: sessionId,
        },
        version: 'v1',
      }
    )

    for await (const item of eventStream) {
      if (item.event === 'on_llm_stream') {
        // Here I send "item.data.chunk.content" to the client as a completion
        // But it sends everything in a single event, whereas I expect something like: Hello,| how |can |I |help |you ? where ‘|’ is an event
      }

      if (item.event === 'on_chain_end') {
       // [End event is sent]
      }
    }
  }
}

Would love to also stream the events properly since we have a setup for multiple llms, providers and its crucial the stream events is working across all these models, it currently doesn't work for claude-3.5-sonnet

[jacoblee93]

Thanks folks, we've implemented and merged in Python already. Will try to get to it this week, stay tuned!