From: Anderson A. Giacomolli Date: Sun, 8 Nov 2026 13:12:00 +0000 (-0300) Subject: Share the request chain of a turn X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/eb05edcf5d575e8a3b1040debfb57423312e8320 Share the request chain of a turn The request chain moves out of the turn loop into Chain: it sends a request, logs the replies, runs the tool calls, and sends their results back until a response has no tool call. A turn builds a chain with the prompt, the tools, and the input of the turn, so its requests do not change. The chain also returns the replies of the final answer, because a subagent needs that text for its report and runs the same chain with its own prompt, toolbox, and log. Issue: b1ac698 --- diff --git a/chain.go b/chain.go new file mode 100644 index 0000000..3ab9fe1 --- /dev/null +++ b/chain.go @@ -0,0 +1,155 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" +) + +// errChainLogWrite marks an error of a log write, which ends the chain. +var errChainLogWrite = errors.New("log write failed") + +// ChainClient sends model calls. +type ChainClient interface { + Create(ctx context.Context, request ModelRequest) (ModelResponse, error) +} + +// Chain is one request chain of a turn or of a subagent. It sends request with +// the input of the first request, runs the tool calls of each response, and +// sends their results back, until a response has no tool call: the final +// answer. The replies, the tool calls, and the results go to the log of tools +// as they happen. The thoughts never do. +// +// Each later request begins with the whole input of the request before it, +// then its output items, then the tool results. The output items hold the +// reasoning with its encrypted content, because the calls do not store it. So +// each request begins with all of the previous one and reads it from the prompt +// cache. The messages of take join each later request after the tool results. +type Chain struct { + client ChainClient + tools *Toolbox + request ModelRequest + take func() []Message +} + +// Run runs the chain from input, the input of its first request, and returns +// the text of the final answer. An error of a log write wraps +// errChainLogWrite. +func (c *Chain) Run(ctx context.Context, input []any) (string, error) { + request := c.request + for request.Turn = "first"; ; request.Turn = "next" { + request.Input = input + response, err := c.client.Create(ctx, request) + if err != nil { + return "", err + } + + var text string + input, text, err = c.handle(ctx, input, response) + if err != nil { + return "", err + } + + if len(input) == 0 { + return text, nil + } + } +} + +// 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 messages of take. After the final answer it returns no input, and the +// replies of the answer as its text. +func (c *Chain) handle(ctx context.Context, input []any, response ModelResponse) ([]any, string, error) { + var outputs []any + var replies []string + for _, raw := range response.Output { + item, err := chainInputItem(raw) + if err != nil { + return nil, "", err + } + + input = append(input, item) + + var call ChainFunctionCall + err = json.Unmarshal(raw, &call) + if err != nil { + return nil, "", fmt.Errorf("decode output item: %w", err) + } + + switch call.Type { + case "message": + text, err := modelOutputText([]json.RawMessage{raw}) + if err != nil { + return nil, "", err + } + + text = strings.TrimSpace(text) + if text == "" { + continue + } + + _, err = c.tools.log.Append("reply", text) + if err != nil { + return nil, "", fmt.Errorf("%w: %w", errChainLogWrite, err) + } + + replies = append(replies, text) + case "function_call": + output, err := c.tools.Call(ctx, call.Name, call.Arguments) + if err != nil { + return nil, "", fmt.Errorf("%w: %w", errChainLogWrite, err) + } + + outputs = append(outputs, ChainFunctionCallOutput{ + Type: "function_call_output", + CallID: call.CallID, + Output: output, + }) + } + } + + if len(outputs) == 0 { + return nil, strings.Join(replies, "\n\n"), nil + } + + input = append(input, outputs...) + for _, message := range c.take() { + input = append(input, modelUserMessage(message.Text)) + } + + return input, "", nil +} + +// ChainFunctionCall holds the fields of a function call output item. +type ChainFunctionCall struct { + Type string `json:"type"` + Name string `json:"name"` + Arguments string `json:"arguments"` + CallID string `json:"call_id"` +} + +// ChainFunctionCallOutput is the input item with the result of a function +// call. +type ChainFunctionCallOutput struct { + Type string `json:"type"` + CallID string `json:"call_id"` + Output string `json:"output"` +} + +// chainInputItem 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 chainInputItem(raw json.RawMessage) (map[string]any, error) { + var item map[string]any + err := json.Unmarshal(raw, &item) + if err != nil { + return nil, fmt.Errorf("decode output item: %w", err) + } + + delete(item, "id") + + return item, nil +} diff --git a/chain_test.go b/chain_test.go new file mode 100644 index 0000000..833f86e --- /dev/null +++ b/chain_test.go @@ -0,0 +1,66 @@ +package main + +import ( + "net/http" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestChainReturnsTheRepliesOfTheFinalAnswer(t *testing.T) { + server := newModelFake(t) + server.reply = func(w http.ResponseWriter, call ModelFakeCall) { + if call.number == 0 { + modelWriteEvents(w, modelCompletedEvent([]any{ + modelMessageItem("Looking."), + modelFunctionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`), + }, 100, 0, 5)) + + return + } + + modelWriteEvents(w, modelCompletedEvent([]any{ + modelMessageItem("It said hi."), + modelMessageItem("Done."), + }, 120, 100, 5)) + } + + memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer()) + require.NoError(t, err) + + t.Cleanup(func() { + require.NoError(t, memory.Close()) + }) + + chain := &Chain{ + client: newModelClient(server.URL, ModelAPIKey(modelTestKey), "agent-key", filepath.Join(t.TempDir(), "usage.jsonl")), + tools: &Toolbox{ + log: memory, + chat: memory, + workdir: t.TempDir(), + }, + request: ModelRequest{ + Model: "gpt-test", + Effort: "low", + }, + take: func() []Message { + return nil + }, + } + + text, err := chain.Run(t.Context(), []any{modelUserMessage("Run echo hi.")}) + require.NoError(t, err) + require.Equal(t, "It said hi.\n\nDone.", text) + require.Equal(t, 2, server.Calls()) + + var kinds []string + for id := range memory.Len() { + message, err := memory.Message(id) + require.NoError(t, err) + + kinds = append(kinds, message.Kind) + } + + require.Equal(t, []string{"reply", "tool", "echo", "reply", "reply"}, kinds) +} diff --git a/turn.go b/turn.go index c6a083b..051c74d 100644 --- a/turn.go +++ b/turn.go @@ -2,21 +2,15 @@ package main import ( "context" - "encoding/json" "errors" - "fmt" "path/filepath" - "strings" "sync" "time" ) -// 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) { +func newTurnLoop(agent *Agent, memory *Memory, client ChainClient, clock func() time.Time, notify func(Event)) (*TurnLoop, error) { layout, err := loadViewLayout(filepath.Join(agent.dir, viewLayoutFileName)) if err != nil { return nil, err @@ -41,11 +35,6 @@ func newTurnLoop(agent *Agent, memory *Memory, client TurnClient, clock func() t 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 ready, busy, idle, // and error. ID is the ID of a logged message. Source names the client that @@ -60,24 +49,17 @@ type Event struct { // 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 the turn runs a Chain whose first request holds the prompt, +// the tools, and the input of turnInput. // // 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 -// hold the reasoning with its encrypted content, because the calls do not -// store it. So each request begins with all of the previous one and reads it -// 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. +// turn reads the view of the turn before from the prompt cache. User messages +// that come during a turn join the next request of its chain after the tool +// results. type TurnLoop struct { agent *Agent memory *Memory - client TurnClient + client ChainClient tools *Toolbox layout *ViewLayout clock func() time.Time @@ -212,7 +194,7 @@ func (l *TurnLoop) turn(ctx context.Context) error { }) err := l.chain(ctx) - if err != nil && !errors.Is(err, errTurnLogWrite) { + if err != nil && !errors.Is(err, errChainLogWrite) { l.notify(Event{ Kind: "error", Text: turnError(ctx, err), @@ -255,29 +237,21 @@ func (l *TurnLoop) chain(ctx context.Context) error { return err } - input := turnInput(view, l.state(), messages) - for turn := "first"; ; turn = "next" { - response, err := l.client.Create(ctx, ModelRequest{ + chain := &Chain{ + client: l.client, + tools: l.tools, + request: ModelRequest{ Model: l.agent.config.Model, Effort: l.agent.config.Effort, Instructions: instructions, Tools: promptTools(), - Input: input, - Turn: turn, - }) - if err != nil { - return err - } + }, + take: l.take, + } - input, err = l.handle(ctx, input, response) - if err != nil { - return err - } + _, err = chain.Run(ctx, turnInput(view, l.state(), messages)) - if len(input) == 0 { - return nil - } - } + return err } // take returns the user messages that wait for a request, and removes them. @@ -297,83 +271,6 @@ 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 *TurnLoop) handle(ctx context.Context, input []any, response ModelResponse) ([]any, error) { - var outputs []any - for _, raw := range response.Output { - item, err := turnInputItem(raw) - if err != nil { - return nil, err - } - - input = append(input, item) - - var call TurnFunctionCall - err = json.Unmarshal(raw, &call) - if err != nil { - return nil, fmt.Errorf("decode output item: %w", err) - } - - switch call.Type { - case "message": - text, err := modelOutputText([]json.RawMessage{raw}) - if err != nil { - return nil, err - } - - text = strings.TrimSpace(text) - if text == "" { - continue - } - - _, err = l.Append("reply", text) - if err != nil { - 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", errTurnLogWrite, err) - } - - outputs = append(outputs, TurnFunctionCallOutput{ - Type: "function_call_output", - CallID: call.CallID, - Output: output, - }) - } - } - - if len(outputs) == 0 { - return nil, nil - } - - input = append(input, outputs...) - for _, message := range l.take() { - input = append(input, modelUserMessage(message.Text)) - } - - return input, nil -} - -// 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"` -} - -// 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"` -} - // turnError returns the text of the error that ended a turn. func turnError(ctx context.Context, err error) string { if ctx.Err() != nil { @@ -400,18 +297,3 @@ func turnInput(view []string, state string, messages []Message) []any { return input } - -// 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 { - return nil, fmt.Errorf("decode output item: %w", err) - } - - delete(item, "id") - - return item, nil -}