Refactor Technical Documentation to Remove Redundant Sections and Streamline Content
- Removed extensive sections on Manager and Executor implementations to focus on high-level concepts and integration points. - Consolidated error handling and global singleton sections for clarity and brevity. - Updated the structure of the document to enhance readability and ensure a more cohesive presentation of the autonomous agent's functionality.
This commit is contained in:
Max
parent
3283bc027d
commit
f6b56ac6cc
1 changed files with 1 additions and 602 deletions
|
|
@ -1017,420 +1017,7 @@ type Dedup interface {
|
|||
|
||||
---
|
||||
|
||||
## 4. Key Implementations
|
||||
|
||||
### 4.1 Manager Implementation
|
||||
|
||||
```go
|
||||
// manager.go
|
||||
package autonomous
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/yaoapp/kun/log"
|
||||
"github.com/yaoapp/yao/agent/autonomous/types"
|
||||
)
|
||||
|
||||
// Ensure manager implements types.Manager
|
||||
var _ types.Manager = (*manager)(nil)
|
||||
|
||||
type manager struct {
|
||||
cache types.Cache
|
||||
scheduler types.Scheduler
|
||||
dedup types.Dedup
|
||||
executor types.Executor
|
||||
|
||||
ticker *time.Ticker
|
||||
tickerMu sync.Mutex
|
||||
stopCh chan struct{}
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// NewManager creates a new manager
|
||||
func NewManager(cfg *types.SchedulerConfig) types.Manager {
|
||||
return &manager{
|
||||
cache: newCache(),
|
||||
scheduler: newScheduler(cfg),
|
||||
dedup: newDedup(),
|
||||
executor: newExecutor(),
|
||||
stopCh: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (m *manager) Start() error {
|
||||
// Load active robots
|
||||
if err := m.cache.LoadAll(context.Background()); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Start scheduler
|
||||
if err := m.scheduler.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Start ticker (every minute)
|
||||
m.ticker = time.NewTicker(time.Minute)
|
||||
m.wg.Add(1)
|
||||
go m.tickLoop()
|
||||
|
||||
log.Info("Autonomous manager started with %d robots", m.cache.Count())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *manager) Stop() error {
|
||||
close(m.stopCh)
|
||||
m.ticker.Stop()
|
||||
m.wg.Wait()
|
||||
return m.scheduler.Stop()
|
||||
}
|
||||
|
||||
func (m *manager) tickLoop() {
|
||||
defer m.wg.Done()
|
||||
for {
|
||||
select {
|
||||
case <-m.stopCh:
|
||||
return
|
||||
case t := <-m.ticker.C:
|
||||
if err := m.Tick(context.Background(), t); err != nil {
|
||||
log.Error("Tick error: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *manager) Tick(ctx context.Context, now time.Time) error {
|
||||
robots := m.cache.List("") // all teams
|
||||
for _, robot := range robots {
|
||||
// Skip if not autonomous or paused
|
||||
if !robot.AutonomousMode || robot.Status == types.RobotPaused {
|
||||
continue
|
||||
}
|
||||
|
||||
// Check if clock trigger is enabled
|
||||
if !robot.Config.Triggers.IsEnabled(types.TriggerClock) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Check if should run now
|
||||
if !m.shouldRun(robot, now) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Check dedup
|
||||
result, err := m.dedup.CheckExecution(ctx, robot.MemberID, types.TriggerClock)
|
||||
if err != nil {
|
||||
log.Warn("Dedup check error for %s: %v", robot.MemberID, err)
|
||||
continue
|
||||
}
|
||||
if result == types.DedupSkip {
|
||||
continue
|
||||
}
|
||||
|
||||
// Submit to scheduler
|
||||
if err := m.scheduler.Submit(ctx, robot, types.TriggerClock, nil); err != nil {
|
||||
log.Warn("Submit error for %s: %v", robot.MemberID, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *manager) shouldRun(robot *types.Robot, now time.Time) bool {
|
||||
cfg := robot.Config.Clock
|
||||
if cfg == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
loc := cfg.GetLocation()
|
||||
now = now.In(loc)
|
||||
|
||||
switch cfg.Mode {
|
||||
case types.ClockTimes:
|
||||
return m.shouldRunTimes(cfg, now)
|
||||
case types.ClockInterval:
|
||||
return m.shouldRunInterval(robot, cfg, now)
|
||||
case types.ClockDaemon:
|
||||
return robot.CanRun() // always run if can
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (m *manager) shouldRunTimes(cfg *types.Clock, now time.Time) bool {
|
||||
// Check day
|
||||
if len(cfg.Days) > 0 && cfg.Days[0] != "*" {
|
||||
dayMatch := false
|
||||
for _, d := range cfg.Days {
|
||||
if d == now.Weekday().String()[:3] {
|
||||
dayMatch = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !dayMatch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Check time (within 1 minute window)
|
||||
nowTime := now.Format("15:04")
|
||||
for _, t := range cfg.Times {
|
||||
if t == nowTime {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (m *manager) shouldRunInterval(robot *types.Robot, cfg *types.Clock, now time.Time) bool {
|
||||
every, err := time.ParseDuration(cfg.Every)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return now.Sub(robot.LastExecution) >= every
|
||||
}
|
||||
```
|
||||
|
||||
### 4.2 Executor Implementation
|
||||
|
||||
```go
|
||||
// executor.go
|
||||
package autonomous
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
gonanoid "github.com/matoous/go-nanoid/v2"
|
||||
"github.com/yaoapp/kun/log"
|
||||
"github.com/yaoapp/yao/agent/autonomous/types"
|
||||
"github.com/yaoapp/yao/job"
|
||||
)
|
||||
|
||||
// Ensure executor implements types.Executor
|
||||
var _ types.Executor = (*executor)(nil)
|
||||
|
||||
type executor struct{}
|
||||
|
||||
func newExecutor() types.Executor {
|
||||
return &executor{}
|
||||
}
|
||||
|
||||
func (e *executor) Execute(ctx context.Context, robot *types.Robot, triggerType types.TriggerType, triggerData interface{}) (*types.Execution, error) {
|
||||
// Create execution
|
||||
exec := &types.Execution{
|
||||
ID: gonanoid.Must(),
|
||||
MemberID: robot.MemberID,
|
||||
TeamID: robot.TeamID,
|
||||
TriggerType: triggerType,
|
||||
TriggerData: triggerData,
|
||||
StartTime: time.Now(),
|
||||
Status: types.ExecRunning,
|
||||
}
|
||||
|
||||
// Create context with timeout
|
||||
timeout := robot.Config.Clock.GetTimeout()
|
||||
exec.ctx, exec.cancel = context.WithTimeout(ctx, timeout)
|
||||
defer exec.cancel()
|
||||
|
||||
// Update robot status
|
||||
robot.IncrRunning()
|
||||
defer robot.DecrRunning()
|
||||
|
||||
// Determine phases to run
|
||||
phases := e.getPhasesToRun(triggerType)
|
||||
|
||||
// Run phases
|
||||
for _, phase := range phases {
|
||||
exec.Phase = phase
|
||||
if err := e.RunPhase(exec.ctx, exec, phase); err != nil {
|
||||
exec.Status = ExecFailed
|
||||
exec.Error = err.Error()
|
||||
e.saveExecution(exec)
|
||||
return exec, err
|
||||
}
|
||||
}
|
||||
|
||||
// Mark completed
|
||||
now := time.Now()
|
||||
exec.EndTime = &now
|
||||
exec.Status = types.ExecCompleted
|
||||
e.saveExecution(exec)
|
||||
|
||||
return exec, nil
|
||||
}
|
||||
|
||||
func (e *executor) getPhasesToRun(triggerType types.TriggerType) []types.Phase {
|
||||
if triggerType == types.TriggerClock {
|
||||
return types.AllPhases // P0 -> P5
|
||||
}
|
||||
// Human/Event: skip P0
|
||||
return []types.Phase{types.PhaseGoals, types.PhaseTasks, types.PhaseRun, types.PhaseDelivery, types.PhaseLearning}
|
||||
}
|
||||
|
||||
func (e *executor) RunPhase(ctx context.Context, exec *types.Execution, phase types.Phase) error {
|
||||
log.Debug("Running phase %s for %s", phase, exec.MemberID)
|
||||
|
||||
switch phase {
|
||||
case types.PhaseInspiration:
|
||||
return e.runInspiration(ctx, exec)
|
||||
case types.PhaseGoals:
|
||||
return e.runGoals(ctx, exec)
|
||||
case types.PhaseTasks:
|
||||
return e.runTasks(ctx, exec)
|
||||
case types.PhaseRun:
|
||||
return e.runExecution(ctx, exec)
|
||||
case types.PhaseDelivery:
|
||||
return e.runDelivery(ctx, exec)
|
||||
case types.PhaseLearning:
|
||||
return e.runLearning(ctx, exec)
|
||||
default:
|
||||
return fmt.Errorf("unknown phase: %s", phase)
|
||||
}
|
||||
}
|
||||
|
||||
func (e *executor) saveExecution(exec *types.Execution) {
|
||||
// Save to job system
|
||||
jobExec := &job.Execution{
|
||||
ExecutionID: exec.ID,
|
||||
JobID: "robot_" + exec.MemberID,
|
||||
Status: string(exec.Status),
|
||||
TriggerCategory: string(exec.TriggerType),
|
||||
}
|
||||
if exec.StartTime.IsZero() == false {
|
||||
jobExec.StartedAt = &exec.StartTime
|
||||
}
|
||||
if exec.EndTime != nil {
|
||||
jobExec.EndedAt = exec.EndTime
|
||||
}
|
||||
job.SaveExecution(jobExec)
|
||||
}
|
||||
```
|
||||
|
||||
### 4.3 Yao Process
|
||||
|
||||
```go
|
||||
// process.go
|
||||
package autonomous
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/yaoapp/gou/process"
|
||||
"github.com/yaoapp/kun/log"
|
||||
"github.com/yaoapp/yao/agent/autonomous/types"
|
||||
)
|
||||
|
||||
func init() {
|
||||
process.Register("autonomous.Execute", processExecute)
|
||||
process.Register("autonomous.Intervene", processIntervene)
|
||||
process.Register("autonomous.HandleEvent", processHandleEvent)
|
||||
process.Register("autonomous.GetStatus", processGetStatus)
|
||||
process.Register("autonomous.Pause", processPause)
|
||||
process.Register("autonomous.Resume", processResume)
|
||||
}
|
||||
|
||||
// processExecute - autonomous.Execute(memberID, triggerType, triggerData)
|
||||
func processExecute(p *process.Process) interface{} {
|
||||
memberID := p.ArgsString(0)
|
||||
triggerType := types.TriggerType(p.ArgsString(1, "clock"))
|
||||
triggerData := p.Args[2]
|
||||
|
||||
mgr := GetManager()
|
||||
robot := mgr.GetRobot("", memberID) // teamID not needed for lookup
|
||||
if robot == nil {
|
||||
return map[string]interface{}{"error": "robot not found"}
|
||||
}
|
||||
|
||||
exec, err := GetExecutor().Execute(context.Background(), robot, triggerType, triggerData)
|
||||
if err != nil {
|
||||
log.Error("Execute error: %v", err)
|
||||
return map[string]interface{}{"error": err.Error()}
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"execution_id": exec.ID,
|
||||
"status": exec.Status,
|
||||
}
|
||||
}
|
||||
|
||||
// processIntervene - autonomous.Intervene(teamID, memberID, action, description, priority)
|
||||
func processIntervene(p *process.Process) interface{} {
|
||||
req := &types.InterveneRequest{
|
||||
TeamID: p.ArgsString(0),
|
||||
MemberID: p.ArgsString(1),
|
||||
Action: types.InterventionAction(p.ArgsString(2)),
|
||||
Description: p.ArgsString(3),
|
||||
Priority: types.Priority(p.ArgsString(4, "normal")),
|
||||
}
|
||||
|
||||
result, err := GetTrigger().Intervene(context.Background(), req)
|
||||
if err != nil {
|
||||
return map[string]interface{}{"error": err.Error()}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// processHandleEvent - autonomous.HandleEvent(memberID, source, eventType, data)
|
||||
func processHandleEvent(p *process.Process) interface{} {
|
||||
req := &types.EventRequest{
|
||||
MemberID: p.ArgsString(0),
|
||||
Source: p.ArgsString(1),
|
||||
EventType: p.ArgsString(2),
|
||||
Data: p.ArgsMap(3),
|
||||
}
|
||||
|
||||
result, err := GetTrigger().HandleEvent(context.Background(), req)
|
||||
if err != nil {
|
||||
return map[string]interface{}{"error": err.Error()}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// processGetStatus - autonomous.GetStatus(teamID, memberID)
|
||||
func processGetStatus(p *process.Process) interface{} {
|
||||
state, err := GetTrigger().GetStatus(
|
||||
context.Background(),
|
||||
p.ArgsString(0),
|
||||
p.ArgsString(1),
|
||||
)
|
||||
if err != nil {
|
||||
return map[string]interface{}{"error": err.Error()}
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
||||
// processPause - autonomous.Pause(teamID, memberID)
|
||||
func processPause(p *process.Process) interface{} {
|
||||
err := GetTrigger().Pause(
|
||||
context.Background(),
|
||||
p.ArgsString(0),
|
||||
p.ArgsString(1),
|
||||
)
|
||||
if err != nil {
|
||||
return map[string]interface{}{"error": err.Error()}
|
||||
}
|
||||
return map[string]interface{}{"success": true}
|
||||
}
|
||||
|
||||
// processResume - autonomous.Resume(teamID, memberID)
|
||||
func processResume(p *process.Process) interface{} {
|
||||
err := GetTrigger().Resume(
|
||||
context.Background(),
|
||||
p.ArgsString(0),
|
||||
p.ArgsString(1),
|
||||
)
|
||||
if err != nil {
|
||||
return map[string]interface{}{"error": err.Error()}
|
||||
}
|
||||
return map[string]interface{}{"success": true}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 5. Errors
|
||||
## 4. Errors
|
||||
|
||||
```go
|
||||
// types/errors.go
|
||||
|
|
@ -1460,191 +1047,3 @@ var (
|
|||
ErrDeliveryFailed = errors.New("delivery failed")
|
||||
)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 6. Global Singletons
|
||||
|
||||
```go
|
||||
// global.go
|
||||
package autonomous
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"github.com/yaoapp/yao/agent/autonomous/types"
|
||||
)
|
||||
|
||||
var (
|
||||
globalManager types.Manager
|
||||
globalTrigger types.Trigger
|
||||
globalExecutor types.Executor
|
||||
globalOnce sync.Once
|
||||
)
|
||||
|
||||
// Init initializes the autonomous system
|
||||
func Init(cfg *types.SchedulerConfig) error {
|
||||
var initErr error
|
||||
globalOnce.Do(func() {
|
||||
mgr := NewManager(cfg)
|
||||
if err := mgr.Start(); err != nil {
|
||||
initErr = err
|
||||
return
|
||||
}
|
||||
globalManager = mgr
|
||||
globalTrigger = newTrigger(mgr)
|
||||
globalExecutor = newExecutor()
|
||||
})
|
||||
return initErr
|
||||
}
|
||||
|
||||
// GetManager returns the global manager
|
||||
func GetManager() types.Manager {
|
||||
return globalManager
|
||||
}
|
||||
|
||||
// GetTrigger returns the global trigger
|
||||
func GetTrigger() types.Trigger {
|
||||
return globalTrigger
|
||||
}
|
||||
|
||||
// GetExecutor returns the global executor
|
||||
func GetExecutor() types.Executor {
|
||||
return globalExecutor
|
||||
}
|
||||
|
||||
// Shutdown stops the autonomous system
|
||||
func Shutdown() error {
|
||||
if globalManager != nil {
|
||||
return globalManager.Stop()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 7. Integration Points
|
||||
|
||||
### 7.1 With Job System
|
||||
|
||||
```go
|
||||
// Job creation on robot create
|
||||
func createRobotJob(robot *types.Robot) error {
|
||||
j, err := job.Once(job.GOROUTINE, map[string]interface{}{
|
||||
"job_id": "robot_" + robot.MemberID,
|
||||
"category_id": "autonomous_robot",
|
||||
"name": robot.DisplayName,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return job.SaveJob(j)
|
||||
}
|
||||
```
|
||||
|
||||
### 7.2 With Assistant
|
||||
|
||||
```go
|
||||
// Call phase agent
|
||||
func callPhaseAgent(ctx context.Context, agentID string, prompt string) (string, error) {
|
||||
ast, err := assistant.Get(agentID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
messages := []chatctx.Message{
|
||||
{Role: "user", Content: prompt},
|
||||
}
|
||||
|
||||
resp, err := ast.Stream(chatctx.New(ctx), messages)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return resp.Content, nil
|
||||
}
|
||||
```
|
||||
|
||||
### 7.3 With Member Model
|
||||
|
||||
```go
|
||||
// Load robot from __yao.member
|
||||
func loadRobotFromMember(memberID string) (*types.Robot, error) {
|
||||
mod := model.Select("__yao.member")
|
||||
data, err := mod.Find(memberID, model.QueryParam{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
robot := &types.Robot{
|
||||
MemberID: data.Get("member_id").(string),
|
||||
TeamID: data.Get("team_id").(string),
|
||||
DisplayName: data.Get("display_name").(string),
|
||||
SystemPrompt: data.Get("system_prompt").(string),
|
||||
Status: types.RobotStatus(data.Get("robot_status").(string)),
|
||||
AutonomousMode: data.Get("autonomous_mode").(bool),
|
||||
}
|
||||
|
||||
// Parse robot_config
|
||||
if cfgData := data.Get("robot_config"); cfgData != nil {
|
||||
var cfg types.Config
|
||||
if err := jsoniter.Unmarshal(cfgData.([]byte), &cfg); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
robot.Config = &cfg
|
||||
}
|
||||
|
||||
return robot, nil
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 8. Testing
|
||||
|
||||
```go
|
||||
// manager_test.go
|
||||
package autonomous
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/yaoapp/yao/agent/autonomous/types"
|
||||
)
|
||||
|
||||
func TestManagerTick(t *testing.T) {
|
||||
mgr := NewManager(&types.SchedulerConfig{Workers: 2})
|
||||
defer mgr.Stop()
|
||||
|
||||
// Add test robot
|
||||
robot := &types.Robot{
|
||||
MemberID: "test_robot",
|
||||
TeamID: "test_team",
|
||||
AutonomousMode: true,
|
||||
Status: types.RobotIdle,
|
||||
Config: &types.Config{
|
||||
Clock: &types.Clock{
|
||||
Mode: types.ClockTimes,
|
||||
Times: []string{"09:00"},
|
||||
Days: []string{"*"},
|
||||
},
|
||||
Identity: &types.Identity{Role: "Test"},
|
||||
},
|
||||
}
|
||||
|
||||
mgr.(*manager).cache.Add(robot)
|
||||
|
||||
// Tick at 09:00
|
||||
now := time.Date(2024, 1, 1, 9, 0, 0, 0, time.Local)
|
||||
err := mgr.Tick(context.Background(), now)
|
||||
if err != nil {
|
||||
t.Fatalf("Tick error: %v", err)
|
||||
}
|
||||
|
||||
// Check execution was submitted
|
||||
// ...
|
||||
}
|
||||
```
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue