cloudwego
[BUG] 使用 StreamToolCallChecker 导致只有全部回答完毕才能在msgReader.Recv()接收
Describe the bug
因为调用的模型会先输出文本，再输出工具，且修改 prompt 无法改变模型的行为，所以按照 StreamToolCallChecker 的指导，加了以下代码（和文档中的一致）：
toolCallChecker := func(ctx context.Context, sr *schema.StreamReader[*schema.Message]) (bool, error) {
defer sr.Close()
for {
msg, err := sr.Recv()
if err != nil {
if errors.Is(err, io.EOF) {
// finish
break
}
return false, err
}
if len(msg.ToolCalls) > 0 {
return true, nil
}
}
return false, nil
}
在 cloudwego/eino v0.4.8 的时候，行为是符合预期的。
但是最近更新到 v0.5.4 之后（cloudwego/eino-ext 也同步升级了），使用 stream 接收，只有等模型的回答全部回答完毕，才能从 stream 的 Recv() 中全部接收。不能像打字机效果一样的返回。
stream 接收代码完全按照 Stream 中的来写的。
移除 StreamToolCallChecker 后，stream.Recv() 符合预期，但是因为模型先输出文本、再输出工具，eino 不支持这种模式，也无法正常使用
Expected behavior
能像打字机效果一样的接收模型回答。
Screenshots
If applicable, add screenshots to help explain your problem.
Version:
v0.5.4
Environment:
Debian 13 go1.25.1
最小可复现代码：
package main
import (
"context"
"errors"
"fmt"
"io"
"log"
"os"
"github.com/cloudwego/eino-ext/components/model/openai"
"github.com/cloudwego/eino/components/tool"
"github.com/cloudwego/eino/compose"
"github.com/cloudwego/eino/flow/agent/react"
"github.com/cloudwego/eino/schema"
)
func main() {
ctx := context.Background()
temperature := float32(0.8)
model, err := openai.NewChatModel(ctx, &openai.ChatModelConfig{
APIKey: os.Getenv("API_KEY"),
Model: "deepseek-chat",
Temperature: &temperature,
BaseURL: "https://api.deepseek.com/v1",
})
if err != nil {
log.Fatal(err)
}
ragent, err := react.NewAgent(ctx, &react.AgentConfig{
ToolCallingModel: model,
ToolsConfig: compose.ToolsNodeConfig{
Tools: []tool.BaseTool{}, // 有没有 tool 都是一样能复现
},
// 注释掉下面这行，将可以正常按照打字机效果输出模型的回答
// 保留下面这行，则需要模型全部回答完，才能全部一起输出
StreamToolCallChecker: toolCallChecker,
})
if err != nil {
log.Fatal(err)
}
sr, err := ragent.Stream(ctx, []*schema.Message{
{
Role: schema.System,
Content: "你是一个智能助手",
},
{
Role: schema.User,
Content: "请简单介绍你自己，100字以内",
},
})
if err != nil {
log.Fatalf("failed to stream: %v", err)
return
}
defer sr.Close() // remember to close the stream
fmt.Printf("\n\n===== start streaming =====\n\n")
for {
msg, err := sr.Recv()
if err != nil {
if errors.Is(err, io.EOF) {
// finish
break
}
// error
fmt.Printf("failed to recv: %v\n", err)
return
}
fmt.Println(msg.Content)
}
fmt.Printf("\n\n===== finished =====\n")
}
func toolCallChecker(ctx context.Context, sr *schema.StreamReader[*schema.Message]) (bool, error) {
defer sr.Close()
for {
msg, err := sr.Recv()
if err != nil {
if errors.Is(err, io.EOF) {
// finish
break
}
return false, err
}
if len(msg.ToolCalls) > 0 {
return true, nil
}
}
return false, nil
}
@cyberxnomad 文档上有提及另一种 toolCallChecker 写法的副作用，可以改用 WithMessageFuture 读取流式 chunk。
@cyberxnomad 文档上有提及另一种 toolCallChecker 写法的副作用，可以改用 WithMessageFuture 读取流式 chunk。
未在文档中找到 WithMessageFuture 的用法
