git.giacomolli.uk - onechat/blob - turn.go

   1 package main
   2
   3 import (
   4         "context"
   5         "errors"
   6         "path/filepath"
   7         "sync"
   8         "time"
   9 )
  10
  11 // newTurnLoop returns the loop of agent with its memory and its subagents. Its
  12 // calls go through client, and notify gets each event. The clock gives the date
  13 // of a turn.
  14 func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client ChainClient, clock func() time.Time, notify func(Event)) (*TurnLoop, error) {
  15         layout, err := loadViewLayout(filepath.Join(agent.dir, viewLayoutFileName))
  16         if err != nil {
  17                 return nil, err
  18         }
  19
  20         l := &TurnLoop{
  21                 agent:  agent,
  22                 memory: memory,
  23                 client: client,
  24                 layout: layout,
  25                 clock:  clock,
  26                 notify: notify,
  27                 wake:   make(chan struct{}, 1),
  28         }
  29
  30         l.tools = &Toolbox{
  31                 log:       l,
  32                 chat:      memory,
  33                 subagents: subagents,
  34                 workdir:   agent.config.Workdir,
  35         }
  36
  37         return l, nil
  38 }
  39
  40 // Event tells the clients what the agent does as it happens. Kind is the kind
  41 // of a logged message (user, reply, tool, echo), or one of ready, busy, idle,
  42 // and error. ID is the ID of a logged message. Source names the client that
  43 // sent a user message, such as terminal or telegram, so that each client can
  44 // show the messages of the others.
  45 type Event struct {
  46         Kind   string `json:"kind"`
  47         ID     int    `json:"id,omitempty"`
  48         Text   string `json:"text,omitempty"`
  49         Source string `json:"source,omitempty"`
  50 }
  51
  52 // TurnLoop runs the turns of one agent. A user message goes to the log when it
  53 // comes, and it wakes the loop. A turn waits until every message before it is
  54 // summarized. Then the turn runs a Chain whose first request holds the prompt,
  55 // the tools, and the input of turnInput.
  56 //
  57 // The view goes out as the messages of a ViewLayout, so the first request of a
  58 // turn reads the view of the turn before from the prompt cache. User messages
  59 // that come during a turn join the next request of its chain after the tool
  60 // results.
  61 type TurnLoop struct {
  62         agent  *Agent
  63         memory *Memory
  64         client ChainClient
  65         tools  *Toolbox
  66         layout *ViewLayout
  67         clock  func() time.Time
  68         notify func(Event)
  69         wake   chan struct{}
  70
  71         mutex   sync.Mutex
  72         pending []Message
  73         cancel  context.CancelFunc
  74 }
  75
  76 // Submit logs a user message from the client source and wakes the loop. The
  77 // message joins the next request: the first one of a new turn, or the next one
  78 // of a running turn.
  79 func (l *TurnLoop) Submit(source, text string) (Message, error) {
  80         message, err := l.record("user", source, text)
  81         if err != nil {
  82                 return Message{}, err
  83         }
  84
  85         l.mutex.Lock()
  86         l.pending = append(l.pending, message)
  87         l.mutex.Unlock()
  88
  89         select {
  90         case l.wake <- struct{}{}:
  91         default:
  92         }
  93
  94         return message, nil
  95 }
  96
  97 // Interrupt ends the running turn. A running tool is killed, and the turn sends
  98 // no further request.
  99 func (l *TurnLoop) Interrupt() bool {
 100         l.mutex.Lock()
 101         defer l.mutex.Unlock()
 102
 103         if l.cancel == nil {
 104                 return false
 105         }
 106
 107         l.cancel()
 108
 109         return true
 110 }
 111
 112 // Busy tells if a turn runs.
 113 func (l *TurnLoop) Busy() bool {
 114         l.mutex.Lock()
 115         defer l.mutex.Unlock()
 116
 117         return l.cancel != nil
 118 }
 119
 120 // Append logs a message and tells the clients. The toolbox logs its calls and
 121 // results through it.
 122 func (l *TurnLoop) Append(kind, text string) (Message, error) {
 123         return l.record(kind, "", text)
 124 }
 125
 126 // Flush makes the log durable.
 127 func (l *TurnLoop) Flush(ctx context.Context) error {
 128         return l.memory.Flush(ctx)
 129 }
 130
 131 // Run runs a turn each time user messages wait, until ctx ends. It returns the
 132 // error of a failed log write, which stops the agent.
 133 func (l *TurnLoop) Run(ctx context.Context) error {
 134         for {
 135                 for l.waiting() {
 136                         err := l.turn(ctx)
 137                         if err != nil {
 138                                 return err
 139                         }
 140                 }
 141
 142                 select {
 143                 case <-l.wake:
 144                 case <-ctx.Done():
 145                         return ctx.Err()
 146                 }
 147         }
 148 }
 149
 150 // record logs a message and tells the clients, with the client source of a user
 151 // message.
 152 func (l *TurnLoop) record(kind, source, text string) (Message, error) {
 153         message, err := l.memory.Append(kind, text)
 154         if err != nil {
 155                 return Message{}, err
 156         }
 157
 158         l.notify(Event{
 159                 Kind:   kind,
 160                 ID:     message.ID,
 161                 Text:   text,
 162                 Source: source,
 163         })
 164
 165         return message, nil
 166 }
 167
 168 // waiting tells if user messages wait for a request.
 169 func (l *TurnLoop) waiting() bool {
 170         l.mutex.Lock()
 171         defer l.mutex.Unlock()
 172
 173         return len(l.pending) > 0
 174 }
 175
 176 // turn runs one turn. A failed or interrupted model call ends the turn and
 177 // tells the clients. The messages of the turn stay in the log and the view, and
 178 // the next turn starts only with a new message, so a lasting failure such as a
 179 // needed login does not repeat. Only a failed log write is an error of turn.
 180 func (l *TurnLoop) turn(ctx context.Context) error {
 181         ctx, cancel := context.WithCancel(ctx)
 182         defer cancel()
 183
 184         l.mutex.Lock()
 185         l.cancel = cancel
 186         l.mutex.Unlock()
 187
 188         defer func() {
 189                 l.mutex.Lock()
 190                 l.cancel = nil
 191                 l.mutex.Unlock()
 192         }()
 193
 194         l.notify(Event{
 195                 Kind: "busy",
 196         })
 197
 198         err := l.chain(ctx)
 199         if err != nil && !errors.Is(err, errChainLogWrite) {
 200                 l.notify(Event{
 201                         Kind: "error",
 202                         Text: turnError(ctx, err),
 203                 })
 204
 205                 err = nil
 206         }
 207
 208         l.notify(Event{
 209                 Kind: "idle",
 210         })
 211
 212         return err
 213 }
 214
 215 // chain runs the request chain of a turn until the final answer.
 216 func (l *TurnLoop) chain(ctx context.Context) error {
 217         messages := l.take()
 218         if len(messages) == 0 {
 219                 return nil
 220         }
 221
 222         err := l.memory.Wait(ctx)
 223         if err != nil {
 224                 return err
 225         }
 226
 227         instructions, err := l.agent.Prompt()
 228         if err != nil {
 229                 return err
 230         }
 231
 232         skip := make(map[int]bool, len(messages))
 233         for _, message := range messages {
 234                 skip[message.ID] = true
 235         }
 236
 237         view, err := l.layout.Split(l.memory.ViewLines(skip))
 238         if err != nil {
 239                 return err
 240         }
 241
 242         chain := &Chain{
 243                 client: l.client,
 244                 tools:  l.tools,
 245                 request: ModelRequest{
 246                         Model:        l.agent.config.Model,
 247                         Effort:       l.agent.config.Effort,
 248                         Instructions: instructions,
 249                         Tools:        promptTools(),
 250                 },
 251                 take: l.take,
 252         }
 253
 254         _, err = chain.Run(ctx, turnInput(view, l.state(), messages))
 255
 256         return err
 257 }
 258
 259 // take returns the user messages that wait for a request, and removes them.
 260 func (l *TurnLoop) take() []Message {
 261         l.mutex.Lock()
 262         defer l.mutex.Unlock()
 263
 264         messages := l.pending
 265         l.pending = nil
 266
 267         return messages
 268 }
 269
 270 // state returns the per-turn state: the end of the view, and the date and time
 271 // of the turn.
 272 func (l *TurnLoop) state() string {
 273         return "</chat>\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00")
 274 }
 275
 276 // turnError returns the text of the error that ended a turn.
 277 func turnError(ctx context.Context, err error) string {
 278         if ctx.Err() != nil {
 279                 return "interrupted"
 280         }
 281
 282         return err.Error()
 283 }
 284
 285 // turnInput returns the input of the first request of a turn, stable parts
 286 // first: a user message for each view message, one with the per-turn state,
 287 // and one for each new message, whole. The view leaves out the leaf lines of
 288 // the new messages.
 289 func turnInput(view []string, state string, messages []Message) []any {
 290         input := make([]any, 0, len(view)+1+len(messages))
 291         for _, text := range view {
 292                 input = append(input, modelUserMessage(text))
 293         }
 294
 295         input = append(input, modelUserMessage(state))
 296         for _, message := range messages {
 297                 input = append(input, modelUserMessage(message.Text))
 298         }
 299
 300         return input
 301 }