X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/e0ab6220b6fd35f7e92a04e4bf149940185713ab..fbee1a3e74297e1f73b3fc99913861b80c96cf87:/turn.go diff --git a/turn.go b/turn.go index 91723d2..c6a083b 100644 --- a/turn.go +++ b/turn.go @@ -11,32 +11,61 @@ import ( "time" ) -// modelClient sends model calls. -type modelClient interface { - Create(ctx context.Context, request Request) (Response, error) +// errTurnLogWrite marks an error of a log write, which ends the loop. +var errTurnLogWrite = errors.New("log write failed") + +// newTurnLoop returns the loop of agent with its memory. Its calls go through +// client, and notify gets each event. The clock gives the date of a turn. +func newTurnLoop(agent *Agent, memory *Memory, client TurnClient, clock func() time.Time, notify func(Event)) (*TurnLoop, error) { + layout, err := loadViewLayout(filepath.Join(agent.dir, viewLayoutFileName)) + if err != nil { + return nil, err + } + + l := &TurnLoop{ + agent: agent, + memory: memory, + client: client, + layout: layout, + clock: clock, + notify: notify, + wake: make(chan struct{}, 1), + } + + l.tools = &Toolbox{ + log: l, + chat: memory, + workdir: agent.config.Workdir, + } + + return l, nil +} + +// TurnClient sends model calls. +type TurnClient interface { + Create(ctx context.Context, request ModelRequest) (ModelResponse, error) } // Event tells the clients what the agent does as it happens. Kind is the kind -// of a logged message (user, reply, tool, echo), or one of busy, idle, and -// error. ID is the ID of a logged message. +// of a logged message (user, reply, tool, echo), or one of ready, busy, idle, +// and error. ID is the ID of a logged message. Source names the client that +// sent a user message, such as terminal or telegram, so that each client can +// show the messages of the others. type Event struct { - Kind string `json:"kind"` - ID int `json:"id,omitempty"` - Text string `json:"text,omitempty"` + Kind string `json:"kind"` + ID int `json:"id,omitempty"` + Text string `json:"text,omitempty"` + Source string `json:"source,omitempty"` } -// loop runs the turns of one agent. A user message goes to the log when it +// TurnLoop runs the turns of one agent. A user message goes to the log when it // comes, and it wakes the loop. A turn waits until every message before it is -// summarized. Then its first request holds the prompt, the tools, and the -// input of turnInput. The loop runs the tool calls of each response and sends -// their results back, until a response has no tool call: the final answer. +// summarized. Then its first request holds the prompt, the tools, and the input +// of turnInput. The loop runs the tool calls of each response and sends their +// results back, until a response has no tool call: the final answer. // -// The view goes out as the messages of a viewLayout. After the turns, when no -// message waits, a warm-up request sends only the prompt, the tools, and the -// view messages, and stops at its first output item. It writes a cache entry -// at the end of the view, where the next turn has a message end, so the first -// request of that turn reads the view from the cache. The warm-up is not -// logged, and its output is not used. +// The view goes out as the messages of a ViewLayout, so the first request of a +// turn reads the view of the turn before from the prompt cache. // // Each later request of a turn begins with the whole input of the request // before it, then its output items, then the tool results. The output items @@ -45,13 +74,12 @@ type Event struct { // from the prompt cache. User messages that come during a turn join the next // request after the tool results. The replies, the tool calls, and the results // go to the log as they happen. The thoughts never do. -type loop struct { +type TurnLoop struct { agent *Agent memory *Memory - client modelClient - tools *toolbox - layout *viewLayout - warmUp bool + client TurnClient + tools *Toolbox + layout *ViewLayout clock func() time.Time notify func(Event) wake chan struct{} @@ -61,54 +89,11 @@ type loop struct { cancel context.CancelFunc } -// functionCall holds the fields of a function call output item. -type functionCall struct { - Type string `json:"type"` - Name string `json:"name"` - Arguments string `json:"arguments"` - CallID string `json:"call_id"` -} - -// functionCallOutput is the input item with the result of a function call. -type functionCallOutput struct { - Type string `json:"type"` - CallID string `json:"call_id"` - Output string `json:"output"` -} - -// newLoop returns the loop of agent with its memory, with the warm-up on. Its -// calls go through client, and notify gets each event. The clock gives the -// date of a turn. -func newLoop(agent *Agent, memory *Memory, client modelClient, clock func() time.Time, notify func(Event)) (*loop, error) { - layout, err := loadLayout(filepath.Join(agent.dir, layoutFileName)) - if err != nil { - return nil, err - } - - l := &loop{ - agent: agent, - memory: memory, - client: client, - layout: layout, - warmUp: true, - clock: clock, - notify: notify, - wake: make(chan struct{}, 1), - } - - l.tools = &toolbox{ - log: l, - chat: memory, - workdir: agent.config.Workdir, - } - - return l, nil -} - -// Submit logs a user message and wakes the loop. The message joins the next -// request: the first one of a new turn, or the next one of a running turn. -func (l *loop) Submit(text string) (Message, error) { - message, err := l.Append("user", text) +// Submit logs a user message from the client source and wakes the loop. The +// message joins the next request: the first one of a new turn, or the next one +// of a running turn. +func (l *TurnLoop) Submit(source, text string) (Message, error) { + message, err := l.record("user", source, text) if err != nil { return Message{}, err } @@ -127,7 +112,7 @@ func (l *loop) Submit(text string) (Message, error) { // Interrupt ends the running turn. A running tool is killed, and the turn sends // no further request. -func (l *loop) Interrupt() bool { +func (l *TurnLoop) Interrupt() bool { l.mutex.Lock() defer l.mutex.Unlock() @@ -140,46 +125,34 @@ func (l *loop) Interrupt() bool { return true } -// Append logs a message and tells the clients. The toolbox logs its calls and -// results through it. -func (l *loop) Append(kind, text string) (Message, error) { - message, err := l.memory.Append(kind, text) - if err != nil { - return Message{}, err - } +// Busy tells if a turn runs. +func (l *TurnLoop) Busy() bool { + l.mutex.Lock() + defer l.mutex.Unlock() - l.notify(Event{ - Kind: kind, - ID: message.ID, - Text: text, - }) + return l.cancel != nil +} - return message, nil +// Append logs a message and tells the clients. The toolbox logs its calls and +// results through it. +func (l *TurnLoop) Append(kind, text string) (Message, error) { + return l.record(kind, "", text) } // Flush makes the log durable. -func (l *loop) Flush(ctx context.Context) error { +func (l *TurnLoop) Flush(ctx context.Context) error { return l.memory.Flush(ctx) } -// Run runs a turn each time user messages wait, and a warm-up after the turns, -// until ctx ends. It returns the error of a failed log write, which stops the -// agent. A failed warm-up only costs the cache of the next turn, so its error -// stays in the usage file. -func (l *loop) Run(ctx context.Context) error { +// Run runs a turn each time user messages wait, until ctx ends. It returns the +// error of a failed log write, which stops the agent. +func (l *TurnLoop) Run(ctx context.Context) error { for { - turned := false for l.waiting() { err := l.turn(ctx) if err != nil { return err } - - turned = true - } - - if turned && l.warmUp { - _ = l.warm(ctx) } select { @@ -190,30 +163,37 @@ func (l *loop) Run(ctx context.Context) error { } } -// waiting tells if user messages wait for a request. -func (l *loop) waiting() bool { - l.mutex.Lock() - defer l.mutex.Unlock() +// record logs a message and tells the clients, with the client source of a user +// message. +func (l *TurnLoop) record(kind, source, text string) (Message, error) { + message, err := l.memory.Append(kind, text) + if err != nil { + return Message{}, err + } - return len(l.pending) > 0 + l.notify(Event{ + Kind: kind, + ID: message.ID, + Text: text, + Source: source, + }) + + return message, nil } -// take returns the user messages that wait for a request, and removes them. -func (l *loop) take() []Message { +// waiting tells if user messages wait for a request. +func (l *TurnLoop) waiting() bool { l.mutex.Lock() defer l.mutex.Unlock() - messages := l.pending - l.pending = nil - - return messages + return len(l.pending) > 0 } // turn runs one turn. A failed or interrupted model call ends the turn and // tells the clients. The messages of the turn stay in the log and the view, and // the next turn starts only with a new message, so a lasting failure such as a // needed login does not repeat. Only a failed log write is an error of turn. -func (l *loop) turn(ctx context.Context) error { +func (l *TurnLoop) turn(ctx context.Context) error { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -232,7 +212,7 @@ func (l *loop) turn(ctx context.Context) error { }) err := l.chain(ctx) - if err != nil && !errors.Is(err, errLogWrite) { + if err != nil && !errors.Is(err, errTurnLogWrite) { l.notify(Event{ Kind: "error", Text: turnError(ctx, err), @@ -248,11 +228,8 @@ func (l *loop) turn(ctx context.Context) error { return err } -// errLogWrite marks an error of a log write, which ends the loop. -var errLogWrite = errors.New("log write failed") - // chain runs the request chain of a turn until the final answer. -func (l *loop) chain(ctx context.Context) error { +func (l *TurnLoop) chain(ctx context.Context) error { messages := l.take() if len(messages) == 0 { return nil @@ -280,11 +257,11 @@ func (l *loop) chain(ctx context.Context) error { input := turnInput(view, l.state(), messages) for turn := "first"; ; turn = "next" { - response, err := l.client.Create(ctx, Request{ + response, err := l.client.Create(ctx, ModelRequest{ Model: l.agent.config.Model, Effort: l.agent.config.Effort, Instructions: instructions, - Tools: agentTools(), + Tools: promptTools(), Input: input, Turn: turn, }) @@ -303,21 +280,38 @@ func (l *loop) chain(ctx context.Context) error { } } +// take returns the user messages that wait for a request, and removes them. +func (l *TurnLoop) take() []Message { + l.mutex.Lock() + defer l.mutex.Unlock() + + messages := l.pending + l.pending = nil + + return messages +} + +// state returns the per-turn state: the end of the view, and the date and time +// of the turn. +func (l *TurnLoop) state() string { + return "\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00") +} + // handle logs the replies of response and runs its tool calls. It returns the // input of the next request: input, the output items, the tool results, and // the user messages that came in the meantime. After the final answer it // returns no input. -func (l *loop) handle(ctx context.Context, input []any, response Response) ([]any, error) { +func (l *TurnLoop) handle(ctx context.Context, input []any, response ModelResponse) ([]any, error) { var outputs []any for _, raw := range response.Output { - item, err := inputItem(raw) + item, err := turnInputItem(raw) if err != nil { return nil, err } input = append(input, item) - var call functionCall + var call TurnFunctionCall err = json.Unmarshal(raw, &call) if err != nil { return nil, fmt.Errorf("decode output item: %w", err) @@ -325,7 +319,7 @@ func (l *loop) handle(ctx context.Context, input []any, response Response) ([]an switch call.Type { case "message": - text, err := outputText([]json.RawMessage{raw}) + text, err := modelOutputText([]json.RawMessage{raw}) if err != nil { return nil, err } @@ -337,15 +331,15 @@ func (l *loop) handle(ctx context.Context, input []any, response Response) ([]an _, err = l.Append("reply", text) if err != nil { - return nil, fmt.Errorf("%w: %w", errLogWrite, err) + return nil, fmt.Errorf("%w: %w", errTurnLogWrite, err) } case "function_call": output, err := l.tools.Call(ctx, call.Name, call.Arguments) if err != nil { - return nil, fmt.Errorf("%w: %w", errLogWrite, err) + return nil, fmt.Errorf("%w: %w", errTurnLogWrite, err) } - outputs = append(outputs, functionCallOutput{ + outputs = append(outputs, TurnFunctionCallOutput{ Type: "function_call_output", CallID: call.CallID, Output: output, @@ -359,47 +353,34 @@ func (l *loop) handle(ctx context.Context, input []any, response Response) ([]an input = append(input, outputs...) for _, message := range l.take() { - input = append(input, userMessage(message.Text)) + input = append(input, modelUserMessage(message.Text)) } return input, nil } -// warm sends the warm-up request, when the builds are done and no message -// waits. -func (l *loop) warm(ctx context.Context) error { - err := l.memory.Wait(ctx) - if err != nil || l.waiting() { - return err - } +// TurnFunctionCall holds the fields of a function call output item. +type TurnFunctionCall struct { + Type string `json:"type"` + Name string `json:"name"` + Arguments string `json:"arguments"` + CallID string `json:"call_id"` +} - instructions, err := l.agent.Prompt() - if err != nil { - return err - } +// TurnFunctionCallOutput is the input item with the result of a function call. +type TurnFunctionCallOutput struct { + Type string `json:"type"` + CallID string `json:"call_id"` + Output string `json:"output"` +} - view, err := l.layout.Split(l.memory.ViewLines(nil)) - if err != nil { - return err +// turnError returns the text of the error that ended a turn. +func turnError(ctx context.Context, err error) string { + if ctx.Err() != nil { + return "interrupted" } - _, err = l.client.Create(ctx, Request{ - Model: l.agent.config.Model, - Effort: l.agent.config.Effort, - Instructions: instructions, - Tools: agentTools(), - Input: warmUpInput(view), - Turn: "warmup", - StopAtOutput: true, - }) - - return err -} - -// state returns the per-turn state: the end of the view, and the date and time -// of the turn. -func (l *loop) state() string { - return "\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00") + return err.Error() } // turnInput returns the input of the first request of a turn, stable parts @@ -407,28 +388,23 @@ func (l *loop) state() string { // and one for each new message, whole. The view leaves out the leaf lines of // the new messages. func turnInput(view []string, state string, messages []Message) []any { - input := warmUpInput(view) - input = append(input, userMessage(state)) - for _, message := range messages { - input = append(input, userMessage(message.Text)) + input := make([]any, 0, len(view)+1+len(messages)) + for _, text := range view { + input = append(input, modelUserMessage(text)) } - return input -} - -// warmUpInput returns a user message for each view message. -func warmUpInput(view []string) []any { - input := make([]any, len(view)) - for n, text := range view { - input[n] = userMessage(text) + input = append(input, modelUserMessage(state)) + for _, message := range messages { + input = append(input, modelUserMessage(message.Text)) } return input } -// inputItem returns an output item as an input item for the next request. The -// calls do not store their items, so the item goes back whole, without its id. -func inputItem(raw json.RawMessage) (map[string]any, error) { +// turnInputItem returns an output item as an input item for the next request. +// The calls do not store their items, so the item goes back whole, without its +// id. +func turnInputItem(raw json.RawMessage) (map[string]any, error) { var item map[string]any err := json.Unmarshal(raw, &item) if err != nil { @@ -439,12 +415,3 @@ func inputItem(raw json.RawMessage) (map[string]any, error) { return item, nil } - -// turnError returns the text of the error that ended a turn. -func turnError(ctx context.Context, err error) string { - if ctx.Err() != nil { - return "interrupted" - } - - return err.Error() -}