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

   1 package main
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "errors"
   7         "fmt"
   8         "path/filepath"
   9         "strings"
  10         "sync"
  11         "time"
  12 )
  13
  14 // modelClient sends model calls.
  15 type modelClient interface {
  16         Create(ctx context.Context, request Request) (Response, error)
  17 }
  18
  19 // Event tells the clients what the agent does as it happens. Kind is the kind
  20 // of a logged message (user, reply, tool, echo), or one of ready, busy, idle,
  21 // and error. ID is the ID of a logged message.
  22 type Event struct {
  23         Kind string `json:"kind"`
  24         ID   int    `json:"id,omitempty"`
  25         Text string `json:"text,omitempty"`
  26 }
  27
  28 // loop runs the turns of one agent. A user message goes to the log when it
  29 // comes, and it wakes the loop. A turn waits until every message before it is
  30 // summarized. Then its first request holds the prompt, the tools, and the
  31 // input of turnInput. The loop runs the tool calls of each response and sends
  32 // their results back, until a response has no tool call: the final answer.
  33 //
  34 // The view goes out as the messages of a viewLayout, so the first request of
  35 // a turn reads the view of the turn before from the prompt cache.
  36 //
  37 // Each later request of a turn begins with the whole input of the request
  38 // before it, then its output items, then the tool results. The output items
  39 // hold the reasoning with its encrypted content, because the calls do not
  40 // store it. So each request begins with all of the previous one and reads it
  41 // from the prompt cache. User messages that come during a turn join the next
  42 // request after the tool results. The replies, the tool calls, and the results
  43 // go to the log as they happen. The thoughts never do.
  44 type loop struct {
  45         agent  *Agent
  46         memory *Memory
  47         client modelClient
  48         tools  *toolbox
  49         layout *viewLayout
  50         clock  func() time.Time
  51         notify func(Event)
  52         wake   chan struct{}
  53
  54         mutex   sync.Mutex
  55         pending []Message
  56         cancel  context.CancelFunc
  57 }
  58
  59 // functionCall holds the fields of a function call output item.
  60 type functionCall struct {
  61         Type      string `json:"type"`
  62         Name      string `json:"name"`
  63         Arguments string `json:"arguments"`
  64         CallID    string `json:"call_id"`
  65 }
  66
  67 // functionCallOutput is the input item with the result of a function call.
  68 type functionCallOutput struct {
  69         Type   string `json:"type"`
  70         CallID string `json:"call_id"`
  71         Output string `json:"output"`
  72 }
  73
  74 // newLoop returns the loop of agent with its memory. Its calls go through
  75 // client, and notify gets each event. The clock gives the date of a turn.
  76 func newLoop(agent *Agent, memory *Memory, client modelClient, clock func() time.Time, notify func(Event)) (*loop, error) {
  77         layout, err := loadLayout(filepath.Join(agent.dir, layoutFileName))
  78         if err != nil {
  79                 return nil, err
  80         }
  81
  82         l := &loop{
  83                 agent:  agent,
  84                 memory: memory,
  85                 client: client,
  86                 layout: layout,
  87                 clock:  clock,
  88                 notify: notify,
  89                 wake:   make(chan struct{}, 1),
  90         }
  91
  92         l.tools = &toolbox{
  93                 log:     l,
  94                 chat:    memory,
  95                 workdir: agent.config.Workdir,
  96         }
  97
  98         return l, nil
  99 }
 100
 101 // Submit logs a user message and wakes the loop. The message joins the next
 102 // request: the first one of a new turn, or the next one of a running turn.
 103 func (l *loop) Submit(text string) (Message, error) {
 104         message, err := l.Append("user", text)
 105         if err != nil {
 106                 return Message{}, err
 107         }
 108
 109         l.mutex.Lock()
 110         l.pending = append(l.pending, message)
 111         l.mutex.Unlock()
 112
 113         select {
 114         case l.wake <- struct{}{}:
 115         default:
 116         }
 117
 118         return message, nil
 119 }
 120
 121 // Interrupt ends the running turn. A running tool is killed, and the turn sends
 122 // no further request.
 123 func (l *loop) Interrupt() bool {
 124         l.mutex.Lock()
 125         defer l.mutex.Unlock()
 126
 127         if l.cancel == nil {
 128                 return false
 129         }
 130
 131         l.cancel()
 132
 133         return true
 134 }
 135
 136 // Busy tells if a turn runs.
 137 func (l *loop) Busy() bool {
 138         l.mutex.Lock()
 139         defer l.mutex.Unlock()
 140
 141         return l.cancel != nil
 142 }
 143
 144 // Append logs a message and tells the clients. The toolbox logs its calls and
 145 // results through it.
 146 func (l *loop) Append(kind, text string) (Message, error) {
 147         message, err := l.memory.Append(kind, text)
 148         if err != nil {
 149                 return Message{}, err
 150         }
 151
 152         l.notify(Event{
 153                 Kind: kind,
 154                 ID:   message.ID,
 155                 Text: text,
 156         })
 157
 158         return message, nil
 159 }
 160
 161 // Flush makes the log durable.
 162 func (l *loop) Flush(ctx context.Context) error {
 163         return l.memory.Flush(ctx)
 164 }
 165
 166 // Run runs a turn each time user messages wait, until ctx ends. It returns the
 167 // error of a failed log write, which stops the agent.
 168 func (l *loop) Run(ctx context.Context) error {
 169         for {
 170                 for l.waiting() {
 171                         err := l.turn(ctx)
 172                         if err != nil {
 173                                 return err
 174                         }
 175                 }
 176
 177                 select {
 178                 case <-l.wake:
 179                 case <-ctx.Done():
 180                         return ctx.Err()
 181                 }
 182         }
 183 }
 184
 185 // waiting tells if user messages wait for a request.
 186 func (l *loop) waiting() bool {
 187         l.mutex.Lock()
 188         defer l.mutex.Unlock()
 189
 190         return len(l.pending) > 0
 191 }
 192
 193 // take returns the user messages that wait for a request, and removes them.
 194 func (l *loop) take() []Message {
 195         l.mutex.Lock()
 196         defer l.mutex.Unlock()
 197
 198         messages := l.pending
 199         l.pending = nil
 200
 201         return messages
 202 }
 203
 204 // turn runs one turn. A failed or interrupted model call ends the turn and
 205 // tells the clients. The messages of the turn stay in the log and the view, and
 206 // the next turn starts only with a new message, so a lasting failure such as a
 207 // needed login does not repeat. Only a failed log write is an error of turn.
 208 func (l *loop) turn(ctx context.Context) error {
 209         ctx, cancel := context.WithCancel(ctx)
 210         defer cancel()
 211
 212         l.mutex.Lock()
 213         l.cancel = cancel
 214         l.mutex.Unlock()
 215
 216         defer func() {
 217                 l.mutex.Lock()
 218                 l.cancel = nil
 219                 l.mutex.Unlock()
 220         }()
 221
 222         l.notify(Event{
 223                 Kind: "busy",
 224         })
 225
 226         err := l.chain(ctx)
 227         if err != nil && !errors.Is(err, errLogWrite) {
 228                 l.notify(Event{
 229                         Kind: "error",
 230                         Text: turnError(ctx, err),
 231                 })
 232
 233                 err = nil
 234         }
 235
 236         l.notify(Event{
 237                 Kind: "idle",
 238         })
 239
 240         return err
 241 }
 242
 243 // errLogWrite marks an error of a log write, which ends the loop.
 244 var errLogWrite = errors.New("log write failed")
 245
 246 // chain runs the request chain of a turn until the final answer.
 247 func (l *loop) chain(ctx context.Context) error {
 248         messages := l.take()
 249         if len(messages) == 0 {
 250                 return nil
 251         }
 252
 253         err := l.memory.Wait(ctx)
 254         if err != nil {
 255                 return err
 256         }
 257
 258         instructions, err := l.agent.Prompt()
 259         if err != nil {
 260                 return err
 261         }
 262
 263         skip := make(map[int]bool, len(messages))
 264         for _, message := range messages {
 265                 skip[message.ID] = true
 266         }
 267
 268         view, err := l.layout.Split(l.memory.ViewLines(skip))
 269         if err != nil {
 270                 return err
 271         }
 272
 273         input := turnInput(view, l.state(), messages)
 274         for turn := "first"; ; turn = "next" {
 275                 response, err := l.client.Create(ctx, Request{
 276                         Model:        l.agent.config.Model,
 277                         Effort:       l.agent.config.Effort,
 278                         Instructions: instructions,
 279                         Tools:        agentTools(),
 280                         Input:        input,
 281                         Turn:         turn,
 282                 })
 283                 if err != nil {
 284                         return err
 285                 }
 286
 287                 input, err = l.handle(ctx, input, response)
 288                 if err != nil {
 289                         return err
 290                 }
 291
 292                 if len(input) == 0 {
 293                         return nil
 294                 }
 295         }
 296 }
 297
 298 // handle logs the replies of response and runs its tool calls. It returns the
 299 // input of the next request: input, the output items, the tool results, and
 300 // the user messages that came in the meantime. After the final answer it
 301 // returns no input.
 302 func (l *loop) handle(ctx context.Context, input []any, response Response) ([]any, error) {
 303         var outputs []any
 304         for _, raw := range response.Output {
 305                 item, err := inputItem(raw)
 306                 if err != nil {
 307                         return nil, err
 308                 }
 309
 310                 input = append(input, item)
 311
 312                 var call functionCall
 313                 err = json.Unmarshal(raw, &call)
 314                 if err != nil {
 315                         return nil, fmt.Errorf("decode output item: %w", err)
 316                 }
 317
 318                 switch call.Type {
 319                 case "message":
 320                         text, err := outputText([]json.RawMessage{raw})
 321                         if err != nil {
 322                                 return nil, err
 323                         }
 324
 325                         text = strings.TrimSpace(text)
 326                         if text == "" {
 327                                 continue
 328                         }
 329
 330                         _, err = l.Append("reply", text)
 331                         if err != nil {
 332                                 return nil, fmt.Errorf("%w: %w", errLogWrite, err)
 333                         }
 334                 case "function_call":
 335                         output, err := l.tools.Call(ctx, call.Name, call.Arguments)
 336                         if err != nil {
 337                                 return nil, fmt.Errorf("%w: %w", errLogWrite, err)
 338                         }
 339
 340                         outputs = append(outputs, functionCallOutput{
 341                                 Type:   "function_call_output",
 342                                 CallID: call.CallID,
 343                                 Output: output,
 344                         })
 345                 }
 346         }
 347
 348         if len(outputs) == 0 {
 349                 return nil, nil
 350         }
 351
 352         input = append(input, outputs...)
 353         for _, message := range l.take() {
 354                 input = append(input, userMessage(message.Text))
 355         }
 356
 357         return input, nil
 358 }
 359
 360 // state returns the per-turn state: the end of the view, and the date and time
 361 // of the turn.
 362 func (l *loop) state() string {
 363         return "</chat>\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00")
 364 }
 365
 366 // turnInput returns the input of the first request of a turn, stable parts
 367 // first: a user message for each view message, one with the per-turn state,
 368 // and one for each new message, whole. The view leaves out the leaf lines of
 369 // the new messages.
 370 func turnInput(view []string, state string, messages []Message) []any {
 371         input := make([]any, 0, len(view)+1+len(messages))
 372         for _, text := range view {
 373                 input = append(input, userMessage(text))
 374         }
 375
 376         input = append(input, userMessage(state))
 377         for _, message := range messages {
 378                 input = append(input, userMessage(message.Text))
 379         }
 380
 381         return input
 382 }
 383
 384 // inputItem returns an output item as an input item for the next request. The
 385 // calls do not store their items, so the item goes back whole, without its id.
 386 func inputItem(raw json.RawMessage) (map[string]any, error) {
 387         var item map[string]any
 388         err := json.Unmarshal(raw, &item)
 389         if err != nil {
 390                 return nil, fmt.Errorf("decode output item: %w", err)
 391         }
 392
 393         delete(item, "id")
 394
 395         return item, nil
 396 }
 397
 398 // turnError returns the text of the error that ended a turn.
 399 func turnError(ctx context.Context, err error) string {
 400         if ctx.Err() != nil {
 401                 return "interrupted"
 402         }
 403
 404         return err.Error()
 405 }