feat(react): dynamic option during execution
Change-Id: I7c67d6b6244c3f37912faf35d64fa4c73bc04358drew/english
shentong.martin
5 months ago
committed by
shentongmartin
parent
82cc9a471d
commit
9835e6abd6
@ -0,0 +1,188 @@
|
||||
# Dynamic Option Modification for ReAct Agent
|
||||
|
||||
This example demonstrates how to dynamically modify `model.Option` during ReAct agent execution. The key challenge is that options passed to `agent.Generate()` or `agent.Stream()` are fixed at call time, but we may want to change options based on the current iteration, previous tool calls, or other runtime conditions.
|
||||
|
||||
## Problem
|
||||
|
||||
When calling a ReAct agent, the option list is passed once and applied to all ChatModel calls during the loop:
|
||||
|
||||
```go
|
||||
agent.Invoke(ctx, messages, opts...) // opts are fixed for all iterations
|
||||
```
|
||||
|
||||
However, you may want to:
|
||||
- Enable/disable extended thinking based on iteration
|
||||
- Change `tool_choice` to force a final answer after N iterations
|
||||
- Modify tool bindings dynamically
|
||||
|
||||
## Solution
|
||||
|
||||
We solve this by:
|
||||
1. **Wrapping the ChatModel** with a `dynamic.ChatModel` that intercepts `Generate()`/`Stream()` calls
|
||||
2. **Using Graph State** via `compose.ProcessState` to persist iteration count across the ReAct loop
|
||||
3. **Wrapping the ReAct Agent** in a parent Graph that provides the state
|
||||
4. **Using MessageFuture** to observe all intermediate results (reasoning, tool calls, tool results)
|
||||
|
||||
### Architecture
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────────────────┐
|
||||
│ Parent Graph (with dynamic.State) │
|
||||
│ ┌───────────────────────────────────────────────────┐ │
|
||||
│ │ ReAct Agent (as sub-graph node) │ │
|
||||
│ │ ┌─────────────────────────────────────────────┐ │ │
|
||||
│ │ │ dynamic.ChatModel │ │ │
|
||||
│ │ │ ├─ Reads iteration from state │ │ │
|
||||
│ │ │ ├─ Calls GetOptionFunc(ctx, input, state) │ │ │
|
||||
│ │ │ ├─ Increments iteration in state │ │ │
|
||||
│ │ │ └─ Calls inner ChatModel with merged opts │ │ │
|
||||
│ │ └─────────────────────────────────────────────┘ │ │
|
||||
│ └───────────────────────────────────────────────────┘ │
|
||||
└─────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Components
|
||||
|
||||
### State (`dynamic/state.go`)
|
||||
|
||||
Holds the iteration counter and optional data for decision making:
|
||||
|
||||
```go
|
||||
type State struct {
|
||||
Iteration int
|
||||
LastToolCalls []*schema.ToolCall
|
||||
CustomData map[string]any
|
||||
}
|
||||
|
||||
type OptionFunc func(ctx context.Context, input []*schema.Message, state *State) []model.Option
|
||||
```
|
||||
|
||||
### ChatModel (`dynamic/model.go`)
|
||||
|
||||
Wraps any `model.BaseChatModel` and uses `compose.ProcessState` to access the graph state:
|
||||
|
||||
```go
|
||||
type ChatModel struct {
|
||||
Model model.BaseChatModel
|
||||
GetOptionFunc OptionFunc
|
||||
}
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
```go
|
||||
// Wrap the ChatModel with dynamic option support
|
||||
dynamicModel := &dynamic.ChatModel{
|
||||
Model: arkChatModel,
|
||||
GetOptionFunc: getDynamicOptions,
|
||||
}
|
||||
|
||||
// Create ReAct agent with the dynamic model
|
||||
rAgent, _ := react.NewAgent(ctx, &react.AgentConfig{
|
||||
ToolCallingModel: dynamicModel,
|
||||
ToolsConfig: toolsConfig,
|
||||
})
|
||||
|
||||
// Create parent graph with local state
|
||||
parentGraph := compose.NewGraph[[]*schema.Message, *schema.Message](
|
||||
compose.WithGenLocalState(func(ctx context.Context) *dynamic.State {
|
||||
return dynamic.NewState()
|
||||
}),
|
||||
)
|
||||
|
||||
// Export and add ReAct agent as sub-graph
|
||||
agentGraph, agentOpts := rAgent.ExportGraph()
|
||||
_ = parentGraph.AddGraphNode("react_agent", agentGraph, agentOpts...)
|
||||
_ = parentGraph.AddEdge(compose.START, "react_agent")
|
||||
_ = parentGraph.AddEdge("react_agent", compose.END)
|
||||
|
||||
runnable, _ := parentGraph.Compile(ctx)
|
||||
|
||||
// Use MessageFuture to observe intermediate results
|
||||
msgFutureOpt, msgFuture := react.WithMessageFuture()
|
||||
|
||||
go func() {
|
||||
// Process intermediate messages in a goroutine
|
||||
iter := msgFuture.GetMessageStreams()
|
||||
for {
|
||||
sr, ok, _ := iter.Next()
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
// Read and print messages...
|
||||
}
|
||||
}()
|
||||
|
||||
// Use Invoke with DesignateNode to pass options to the sub-graph
|
||||
runnable.Invoke(ctx, messages, agent.GetComposeOptions(msgFutureOpt)[0].DesignateNode("react_agent"))
|
||||
```
|
||||
|
||||
## Example: Dynamic Option Function
|
||||
|
||||
```go
|
||||
func getDynamicOptions(ctx context.Context, input []*schema.Message, state *dynamic.State) []model.Option {
|
||||
var opts []model.Option
|
||||
|
||||
// Control thinking mode based on iteration
|
||||
if state.Iteration >= 1 {
|
||||
opts = append(opts, ark.WithThinking(&arkModel.Thinking{
|
||||
Type: arkModel.ThinkingTypeDisabled,
|
||||
}))
|
||||
}
|
||||
|
||||
// Force final answer after first iteration
|
||||
if state.Iteration >= 1 {
|
||||
opts = append(opts, model.WithToolChoice(schema.ToolChoiceForbidden))
|
||||
opts = append(opts, model.WithTools([]*schema.ToolInfo{}))
|
||||
} else {
|
||||
opts = append(opts, model.WithToolChoice(schema.ToolChoiceAllowed))
|
||||
// Bind tools for first iteration
|
||||
opts = append(opts, model.WithTools(toolInfos))
|
||||
}
|
||||
|
||||
return opts
|
||||
}
|
||||
```
|
||||
|
||||
## Observing Intermediate Results with MessageFuture
|
||||
|
||||
The `react.WithMessageFuture()` function returns an option and a `MessageFuture` interface that allows you to observe all intermediate messages during agent execution:
|
||||
|
||||
- **Reasoning Content**: The model's thinking process (`msg.ReasoningContent`)
|
||||
- **Tool Calls**: Function calls made by the model (`msg.ToolCalls`)
|
||||
- **Tool Results**: Results returned from tool execution (`msg.Role == schema.Tool`)
|
||||
- **Assistant Messages**: Text responses from the model
|
||||
|
||||
**Note**: When using `Invoke` instead of `Stream`, you must use `DesignateNode` to pass the MessageFuture option to the correct sub-graph node:
|
||||
|
||||
```go
|
||||
runnable.Invoke(ctx, messages, agent.GetComposeOptions(msgFutureOpt)[0].DesignateNode("react_agent"))
|
||||
```
|
||||
|
||||
## Quick Start
|
||||
|
||||
Environment variables:
|
||||
- `ARK_API_KEY`
|
||||
- `ARK_MODEL_NAME`
|
||||
|
||||
Build and run:
|
||||
|
||||
```bash
|
||||
cd flow/agent/react/dynamic_option_example
|
||||
go build -o dynamic_option_example .
|
||||
./dynamic_option_example
|
||||
```
|
||||
|
||||
## Key Design Decisions
|
||||
|
||||
1. **Graph State over Context**: Context values don't propagate back from `Generate()`, so we use `compose.ProcessState` to persist state across iterations.
|
||||
|
||||
2. **Wrapper Pattern**: Following the decorator pattern used elsewhere in the codebase.
|
||||
|
||||
3. **Simple Function Type**: Using `OptionFunc` instead of an interface keeps the API simple and easy to understand.
|
||||
|
||||
4. **Parent Graph**: The ReAct agent is wrapped as a sub-graph node in a parent graph that provides the state.
|
||||
|
||||
5. **MessageFuture for Observability**: Using `react.WithMessageFuture()` to capture and display all intermediate results including reasoning, tool calls, and tool results.
|
||||
|
||||
6. **Invoke with DesignateNode**: When using `Invoke` instead of `Stream`, use `DesignateNode` to ensure options are passed to the correct sub-graph node.
|
||||
@ -0,0 +1,146 @@
|
||||
/*
|
||||
* Copyright 2025 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 dynamic
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/cloudwego/eino/components"
|
||||
"github.com/cloudwego/eino/components/model"
|
||||
"github.com/cloudwego/eino/compose"
|
||||
"github.com/cloudwego/eino/schema"
|
||||
)
|
||||
|
||||
// ChatModel wraps a BaseChatModel and enables dynamic option modification.
|
||||
// Before each Generate() or Stream() call, it:
|
||||
// 1. Reads the current iteration state from the parent graph via compose.ProcessState
|
||||
// 2. Calls GetOptionFunc to get dynamic options based on the current state
|
||||
// 3. Increments the iteration counter
|
||||
// 4. Merges dynamic options with any static options and calls the inner model
|
||||
type ChatModel struct {
|
||||
// Model is the underlying ChatModel to wrap
|
||||
Model model.BaseChatModel
|
||||
|
||||
// GetOptionFunc is called before each Generate()/Stream() call to get dynamic options
|
||||
GetOptionFunc OptionFunc
|
||||
}
|
||||
|
||||
// Generate implements model.BaseChatModel.
|
||||
// It reads state, calls GetOptionFunc, increments iteration, and delegates to the inner model.
|
||||
func (d *ChatModel) Generate(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.Message, error) {
|
||||
var dynamicOpts []model.Option
|
||||
|
||||
// Access the parent graph's state via compose.ProcessState.
|
||||
// This is the key mechanism that allows state to persist across ReAct loop iterations.
|
||||
// We are accessing parent graph's state here. Require eino version v0.7.11+
|
||||
err := compose.ProcessState[*State](ctx, func(_ context.Context, state *State) error {
|
||||
// Small delay to ensure log ordering (optional, for demo purposes)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
fmt.Printf("\n==================== Iteration %d ====================\n", state.Iteration)
|
||||
|
||||
// Get dynamic options based on current state
|
||||
dynamicOpts = d.GetOptionFunc(ctx, input, state)
|
||||
|
||||
// Increment iteration for next call
|
||||
state.Iteration++
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
// If state access fails (e.g., not running in a graph), use no dynamic options
|
||||
dynamicOpts = nil
|
||||
}
|
||||
|
||||
// Merge dynamic options with static options (dynamic options take precedence)
|
||||
mergedOpts := append(dynamicOpts, opts...)
|
||||
resp, err := d.Model.Generate(ctx, input, mergedOpts...)
|
||||
|
||||
// Store tool calls in state for potential use in next iteration's decision
|
||||
if err == nil && resp != nil && len(resp.ToolCalls) > 0 {
|
||||
_ = compose.ProcessState[*State](ctx, func(_ context.Context, state *State) error {
|
||||
toolCalls := make([]*schema.ToolCall, len(resp.ToolCalls))
|
||||
for i := range resp.ToolCalls {
|
||||
toolCalls[i] = &resp.ToolCalls[i]
|
||||
}
|
||||
state.LastToolCalls = toolCalls
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
return resp, err
|
||||
}
|
||||
|
||||
// Stream implements model.BaseChatModel.
|
||||
// Same logic as Generate but returns a stream reader.
|
||||
func (d *ChatModel) Stream(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.StreamReader[*schema.Message], error) {
|
||||
var dynamicOpts []model.Option
|
||||
|
||||
// We are accessing parent graph's state here. Require eino version v0.7.11+
|
||||
err := compose.ProcessState[*State](ctx, func(_ context.Context, state *State) error {
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
fmt.Printf("\n==================== Iteration %d ====================\n", state.Iteration)
|
||||
dynamicOpts = d.GetOptionFunc(ctx, input, state)
|
||||
state.Iteration++
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
dynamicOpts = nil
|
||||
}
|
||||
|
||||
mergedOpts := append(dynamicOpts, opts...)
|
||||
return d.Model.Stream(ctx, input, mergedOpts...)
|
||||
}
|
||||
|
||||
// WithTools implements model.ToolCallingChatModel.
|
||||
// It creates a new ChatModel wrapping the result of the inner model's WithTools.
|
||||
func (d *ChatModel) WithTools(tools []*schema.ToolInfo) (model.ToolCallingChatModel, error) {
|
||||
tcm, ok := d.Model.(model.ToolCallingChatModel)
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
newModel, err := tcm.WithTools(tools)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ChatModel{
|
||||
Model: newModel,
|
||||
GetOptionFunc: d.GetOptionFunc,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// IsCallbacksEnabled implements components.Checker.
|
||||
// Delegates to the inner model if it implements Checker.
|
||||
func (d *ChatModel) IsCallbacksEnabled() bool {
|
||||
checker, ok := d.Model.(components.Checker)
|
||||
if ok {
|
||||
return checker.IsCallbacksEnabled()
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// GetType returns the type name for this component.
|
||||
func (d *ChatModel) GetType() string {
|
||||
return "DynamicChatModel"
|
||||
}
|
||||
|
||||
// Compile-time interface checks
|
||||
var (
|
||||
_ model.BaseChatModel = (*ChatModel)(nil)
|
||||
_ model.ToolCallingChatModel = (*ChatModel)(nil)
|
||||
_ components.Checker = (*ChatModel)(nil)
|
||||
)
|
||||
@ -0,0 +1,65 @@
|
||||
/*
|
||||
* Copyright 2025 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 dynamic
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/cloudwego/eino/components/model"
|
||||
"github.com/cloudwego/eino/schema"
|
||||
)
|
||||
|
||||
// State holds the iteration state that persists across ReAct loop iterations.
|
||||
// It is stored in the parent graph's local state and accessed via compose.ProcessState.
|
||||
type State struct {
|
||||
// Iteration is the current iteration number (0-indexed).
|
||||
// It is incremented after each ChatModel.Generate() call.
|
||||
Iteration int
|
||||
|
||||
// LastToolCalls stores the tool calls from the previous iteration.
|
||||
// This can be used to make decisions based on what tools were called.
|
||||
LastToolCalls []*schema.ToolCall
|
||||
|
||||
// CustomData allows storing arbitrary data for custom decision logic.
|
||||
CustomData map[string]any
|
||||
}
|
||||
|
||||
func init() {
|
||||
// Register the State type for serialization support in compose graphs.
|
||||
schema.RegisterName[State]("DynamicOptionState")
|
||||
}
|
||||
|
||||
// NewState creates a new State with default values.
|
||||
func NewState() *State {
|
||||
return &State{
|
||||
Iteration: 0,
|
||||
CustomData: make(map[string]any),
|
||||
}
|
||||
}
|
||||
|
||||
// OptionFunc is the function signature for dynamic option generation.
|
||||
// It is called before each ChatModel.Generate() call and returns options
|
||||
// to be merged with any static options passed to the agent.
|
||||
//
|
||||
// Parameters:
|
||||
// - ctx: The context from the current request
|
||||
// - input: The input messages being sent to the ChatModel
|
||||
// - state: The current iteration state (can be modified)
|
||||
//
|
||||
// Returns:
|
||||
// - A slice of model.Option to be applied to this ChatModel call
|
||||
type OptionFunc func(ctx context.Context, input []*schema.Message, state *State) []model.Option
|
||||
Loading…
Reference in New Issue