Closed Moe03 closed 1 month ago
Hey @Moe03! I'm here to help you with any bugs, questions, or contributions. Let's tackle this together while we wait for a human maintainer.
To enable proper streaming of events in the @langchain/anthropic
package, you can utilize the stream
method to yield chunks of data as they are generated. This method ensures that events are streamed in real-time rather than waiting for the entire tool and response to finish.
Here is an example of how you can use the stream
method to achieve this:
import { ChatAnthropic } from "@langchain/anthropic";
import { AgentExecutor, createToolCallingAgent } from "langchain/agents";
import { ChatPromptTemplate, MessagesPlaceholder } from "@langchain/core/prompts";
import { HumanMessage, SystemMessage } from "@langchain/core/messages";
import { DynamicStructuredTool } from "@langchain/core/tools";
import { z } from "zod";
async function main() {
const defaultTools = [
new DynamicStructuredTool({
name: "GetWeather",
description: "Return the weather.",
schema: z.object({
query: z.string().describe("The city to return weather for."),
}),
func: async (input) => {
console.log(`STARTING TOOL: GetWeather`);
return 'sunny';
},
}),
];
const chatPrompt = ChatPromptTemplate.fromMessages([
new MessagesPlaceholder("messages"),
new MessagesPlaceholder("agent_scratchpad"),
]);
const messagesHistory = [
new SystemMessage("you are a helpful assistant!"),
new HumanMessage("Whats the weather like in Cairo?")
];
const llm = new ChatAnthropic({
apiKey: ANTHROPIC_API_KEY,
model: "claude-3-5-sonnet-20240620",
streaming: true, // Enable streaming
});
const agent = await createToolCallingAgent({
llm,
tools: defaultTools,
prompt: chatPrompt,
});
const agentExec = new AgentExecutor({
agent,
tools: defaultTools,
}).withConfig({ runName: "Agent" });
let llmStream = await agentExec.streamEvents(
{
messages: messagesHistory,
},
{
version: "v1"
}
);
for await (const event of llmStream) {
const eventType = event.event;
if (eventType === "on_chain_start") {
if (event.name === "Agent") {
console.log("\n-----");
console.log(`Starting agent: ${event.name} with input: ${JSON.stringify(event.data.input)}`);
}
} else if (eventType === "on_chain_end") {
if (event?.name === "Agent") {
console.log("\n-----");
console.log(`Finished agent: ${event?.name}\n`);
console.log(`Agent output was: ${event?.data?.output}`);
console.log("\n-----");
}
} else if (eventType === "on_llm_stream" || eventType === "on_chat_model_stream" || eventType === "on_chain_stream") {
console.log("\n-----");
const content = event?.data?.chunk?.output || event.data?.chunk?.message?.content || event.data?.chunk?.content;
if (content !== undefined && content !== "") {
console.log(`| ${content}`);
}
} else if (eventType === "on_tool_start") {
console.log("\n-----");
console.log(`Starting tool: ${event.name} with inputs: ${event.data.input}`);
} else if (eventType === "on_tool_end") {
console.log("\n-----");
console.log(`Finished tool: ${event.name}\n`);
console.log(`Tool output was: ${event.data.output}`);
}
}
}
main();
This example ensures that chunks of data are yielded and events are sent in real-time, improving the performance of real-time applications [1][2][3].
To continue talking to Dosu, mention @dosu.
Tracking in #6056
This should now be fixed!
I ran into the same problem and after upgrading to @langchain/anthropic v0.2.6, handleLLMNewToken
is still not triggering for me, unfortunately.
@bracesproul can you have a look?
@bracesproul Thanks in advance!
I think I've found the lines at issue in langchain-anthropic/src/chat_model.ts. The problem is with the extractToken
function -
Basically, extractToken
expects a string but when tools are used it's being passed an object instead.
Here's the step-by-step: First, the code sets coerceContentToString to always be false whenever a tool is used (which seems good and correct).
// this will _always_ return false if a tool is being used
const coerceContentToString = !_toolsInParams({
...params,
...formattedMessages,
stream: false,
});
Then, the streamed tool chunk content is set as the following object in _makeMessageChunkFromAnthropicEvent
:
} else if (
data.type === "content_block_delta" &&
data.delta.type === "input_json_delta"
) {
return {
chunk: new AIMessageChunk({
content: fields.coerceContentToString
? ""
: [
{
index: data.index,
input: data.delta.partial_json, // this input field is where the streamed token is
type: data.delta.type,
},
],
additional_kwargs: {},
}),
usageData: usageDataCopy,
};
}
The next step is where the issue lies. Basically, we have the extractToken
function that's being passed the chunk content. However, it expects the content
field to always be a string, whereas in this case it's an object (the one defined above)! Therefore, because it can't parse an object, it coerces the token value to always be undefined.
const token = extractToken(chunk); // this will always be undefined, as it does not know how to handle the chunk content as an object
Because token is undefined, the callback is never triggered:
if (token) {
await runManager?.handleLLMNewToken(token); // this is unreachable
}
In reality, though, if extractToken
knew how to parse it, the token can be found in chunk.content[0].input
.
For example:
[ { index: 0, input: 'opportu', type: 'input_json_delta' } ]
Is this helpful in terms of tracking it down?
Hey @snlamm I believe this PR should account for the edge case you're mentioning here https://github.com/langchain-ai/langchainjs/pull/6179
Let me know if you're looking for something else.
@bracesproul that looks great! Looks like it accounts for it 🙏 (added a comment on the PR itself re: making sure it's handling the needed type)
@bracesproul Ah looks like the new implementation introduces it's own bug 😅 . Opened a 1-liner to fix at https://github.com/langchain-ai/langchainjs/pull/6183
Can confirm it works now - thanks again!
Checked other resources
Example Code
The following is a more advanced example but it is to showcase the full problem, I've also implemented a fix for it in a pr.
We basically needed tools to stream properly on @langchain/anthropic package to enable streaming events across any other langchain method like the agentExecutor I'm using here.
Error Message and Stack Trace (if applicable)
There are no errors but it doesn't stream when it should.
Description
We basically needed tools to stream properly on @langchain/anthropic package to enable streaming events across any other langchain method like the agentExecutor.
Currenlty it doesn't stream and instead wait for the whole tool + response to finish then outputs the response which takes a really long time and is bad UX in realtime apps.
System Info
Node version 18.17.1 yarn version 1.22.22 platform windows
yarn info langchain: (result was too big)