* Copyright 2024 CloudWeGo Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package utils
import (
"context"
"fmt"
"github.com/bytedance/sonic"
"github.com/cloudwego/eino/components/tool"
"github.com/cloudwego/eino/internal/generic"
"github.com/cloudwego/eino/schema"
)
type StreamFunc[T, D any] func(ctx context.Context, input T) (output *schema.StreamReader[D], err error)
type OptionableStreamFunc[T, D any] func(ctx context.Context, input T, opts ...tool.Option) (output *schema.StreamReader[D], err error)
func InferStreamTool[T, D any](toolName, toolDesc string, s StreamFunc[T, D], opts ...Option) (tool.StreamableTool, error) {
ti, err := goStruct2ToolInfo[T](toolName, toolDesc, opts...)
if err != nil {
return nil, err
}
return NewStreamTool(ti, s, opts...), nil
}
func InferOptionableStreamTool[T, D any](toolName, toolDesc string, s OptionableStreamFunc[T, D], opts ...Option) (tool.StreamableTool, error) {
ti, err := goStruct2ToolInfo[T](toolName, toolDesc, opts...)
if err != nil {
return nil, err
}
return newOptionableStreamTool(ti, s, opts...), nil
}
func NewStreamTool[T, D any](desc *schema.ToolInfo, s StreamFunc[T, D], opts ...Option) tool.StreamableTool {
return newOptionableStreamTool(desc,
func(ctx context.Context, input T, _ ...tool.Option) (output *schema.StreamReader[D], err error) {
return s(ctx, input)
},
opts...)
}
func newOptionableStreamTool[T, D any](desc *schema.ToolInfo, s OptionableStreamFunc[T, D], opts ...Option) tool.StreamableTool {
to := getToolOptions(opts...)
return &streamableTool[T, D]{
info: desc,
um: to.um,
m: to.m,
Fn: s,
}
}
type streamableTool[T, D any] struct {
info *schema.ToolInfo
um UnmarshalArguments
m MarshalOutput
Fn OptionableStreamFunc[T, D]
}
func (s *streamableTool[T, D]) Info(ctx context.Context) (*schema.ToolInfo, error) {
return s.info, nil
}
func (s *streamableTool[T, D]) StreamableRun(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (
outStream *schema.StreamReader[string], err error) {
var inst T
if s.um != nil {
var val any
val, err = s.um(ctx, argumentsInJSON)
if err != nil {
return nil, fmt.Errorf("[LocalStreamFunc] failed to unmarshal arguments, toolName=%s, err=%w", s.getToolName(), err)
}
gt, ok := val.(T)
if !ok {
return nil, fmt.Errorf("[LocalStreamFunc] type err, toolName=%s, expected=%T, given=%T", s.getToolName(), inst, val)
}
inst = gt
} else {
inst = generic.NewInstance[T]()
err = sonic.UnmarshalString(argumentsInJSON, &inst)
if err != nil {
return nil, fmt.Errorf("[LocalStreamFunc] failed to unmarshal arguments in json, toolName=%s, err=%w", s.getToolName(), err)
}
}
streamD, err := s.Fn(ctx, inst, opts...)
if err != nil {
return nil, err
}
outStream = schema.StreamReaderWithConvert(streamD, func(d D) (string, error) {
var out string
var e error
if s.m != nil {
out, e = s.m(ctx, d)
if e != nil {
return "", fmt.Errorf("[LocalStreamFunc] failed to marshal output, toolName=%s, err=%w", s.getToolName(), e)
}
} else {
out, e = marshalString(d)
if e != nil {
return "", fmt.Errorf("[LocalStreamFunc] failed to marshal output in json, toolName=%s, err=%w", s.getToolName(), e)
}
}
return out, nil
})
return outStream, nil
}
func (s *streamableTool[T, D]) GetType() string {
return snakeToCamel(s.getToolName())
}
func (s *streamableTool[T, D]) getToolName() string {
if s.info == nil {
return ""
}
return s.info.Name
}
type EnhancedStreamFunc[T any] func(ctx context.Context, input T) (output *schema.StreamReader[*schema.ToolResult], err error)
type OptionableEnhancedStreamFunc[T any] func(ctx context.Context, input T, opts ...tool.Option) (output *schema.StreamReader[*schema.ToolResult], err error)
func InferEnhancedStreamTool[T any](toolName, toolDesc string, s EnhancedStreamFunc[T], opts ...Option) (tool.EnhancedStreamableTool, error) {
ti, err := goStruct2ToolInfo[T](toolName, toolDesc, opts...)
if err != nil {
return nil, err
}
return NewEnhancedStreamTool(ti, s, opts...), nil
}
func InferOptionableEnhancedStreamTool[T any](toolName, toolDesc string, s OptionableEnhancedStreamFunc[T], opts ...Option) (tool.EnhancedStreamableTool, error) {
ti, err := goStruct2ToolInfo[T](toolName, toolDesc, opts...)
if err != nil {
return nil, err
}
return newOptionableEnhancedStreamTool(ti, s, opts...), nil
}
func NewEnhancedStreamTool[T any](desc *schema.ToolInfo, s EnhancedStreamFunc[T], opts ...Option) tool.EnhancedStreamableTool {
return newOptionableEnhancedStreamTool(desc,
func(ctx context.Context, input T, _ ...tool.Option) (output *schema.StreamReader[*schema.ToolResult], err error) {
return s(ctx, input)
},
opts...)
}
func newOptionableEnhancedStreamTool[T any](desc *schema.ToolInfo, s OptionableEnhancedStreamFunc[T], opts ...Option) tool.EnhancedStreamableTool {
to := getToolOptions(opts...)
return &enhancedStreamableTool[T]{
info: desc,
um: to.um,
Fn: s,
}
}
type enhancedStreamableTool[T any] struct {
info *schema.ToolInfo
um UnmarshalArguments
Fn OptionableEnhancedStreamFunc[T]
}
func (s *enhancedStreamableTool[T]) Info(ctx context.Context) (*schema.ToolInfo, error) {
return s.info, nil
}
func (s *enhancedStreamableTool[T]) StreamableRun(ctx context.Context, toolArgument *schema.ToolArgument, opts ...tool.Option) (
outStream *schema.StreamReader[*schema.ToolResult], err error) {
var inst T
if s.um != nil {
var val any
val, err = s.um(ctx, toolArgument.Text)
if err != nil {
return nil, fmt.Errorf("[EnhancedLocalStreamFunc] failed to unmarshal arguments, toolName=%s, err=%w", s.getToolName(), err)
}
gt, ok := val.(T)
if !ok {
return nil, fmt.Errorf("[EnhancedLocalStreamFunc] type err, toolName=%s, expected=%T, given=%T", s.getToolName(), inst, val)
}
inst = gt
} else {
inst = generic.NewInstance[T]()
err = sonic.UnmarshalString(toolArgument.Text, &inst)
if err != nil {
return nil, fmt.Errorf("[EnhancedLocalStreamFunc] failed to unmarshal arguments in json, toolName=%s, err=%w", s.getToolName(), err)
}
}
return s.Fn(ctx, inst, opts...)
}
func (s *enhancedStreamableTool[T]) GetType() string {
return snakeToCamel(s.getToolName())
}
func (s *enhancedStreamableTool[T]) getToolName() string {
if s.info == nil {
return ""
}
return s.info.Name
}