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