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