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