Avoid intermediate conversation history copies
This commit is contained in:
@@ -32,7 +32,8 @@ The runner has no durable run or session store.
|
|||||||
|
|
||||||
Captured exchanges are immutable parent-linked domain values. Each holds only
|
Captured exchanges are immutable parent-linked domain values. Each holds only
|
||||||
its new rendered suffix, final assistant message, and run identities; flattening
|
its new rendered suffix, final assistant message, and run identities; flattening
|
||||||
iteratively makes a detached history for preparation or inspection. Preparation
|
iteratively fills a single detached message buffer for preparation or projects
|
||||||
|
directly into public inspection values. Preparation
|
||||||
prepends this history to newly rendered definition messages and appended caller
|
prepends this history to newly rendered definition messages and appended caller
|
||||||
messages before computing the rendered hash. Follow-up selection and credentials
|
messages before computing the rendered hash. Follow-up selection and credentials
|
||||||
remain those of the new request. Prepared execution retains capture and parent
|
remain those of the new request. Prepared execution retains capture and parent
|
||||||
|
|||||||
@@ -91,7 +91,12 @@ does not require changing runtime behavior or duplicating existing tests.
|
|||||||
|
|
||||||
## 2. Avoid Intermediate History Copies During Assembly And Inspection
|
## 2. Avoid Intermediate History Copies During Assembly And Inspection
|
||||||
|
|
||||||
**Priority:** P3. **Status:** For consideration. **Plan stages:** 2 and 5.
|
**Priority:** P3. **Status:** Complete. **Plan stages:** 2 and 5.
|
||||||
|
|
||||||
|
**Completion:** Preparation now fills one history-plus-suffix buffer. Inspection
|
||||||
|
projects indexed history directly into public messages, cloning exposed cache
|
||||||
|
metadata once. The parent-linked representation and public types are unchanged.
|
||||||
|
Focused race-enabled history, continuation, preparation, and ownership tests passed.
|
||||||
|
|
||||||
**Locations:** `completePreparationWithStructuredOutput` in
|
**Locations:** `completePreparationWithStructuredOutput` in
|
||||||
[the runner](../../internal/usecase/runner.go), `Exchange.Messages` in
|
[the runner](../../internal/usecase/runner.go), `Exchange.Messages` in
|
||||||
|
|||||||
@@ -33,7 +33,11 @@ func (e *Exchange) Details() ExchangeDetails {
|
|||||||
if e == nil || e.internal == nil {
|
if e == nil || e.internal == nil {
|
||||||
return ExchangeDetails{}
|
return ExchangeDetails{}
|
||||||
}
|
}
|
||||||
return ExchangeDetails{RunID: e.internal.RunID(), ParentRunID: e.internal.ParentRunID(), Messages: fromDomainRenderedMessages(e.internal.Messages())}
|
messages := make([]RenderedMessage, e.internal.MessageCount())
|
||||||
|
e.internal.VisitMessages(func(index int, message domain.RenderedMessage) {
|
||||||
|
messages[index] = RenderedMessage{Role: message.Role, Content: message.Content, CacheControl: fromDomainCacheControl(message.CacheControl)}
|
||||||
|
})
|
||||||
|
return ExchangeDetails{RunID: e.internal.RunID(), ParentRunID: e.internal.ParentRunID(), Messages: messages}
|
||||||
}
|
}
|
||||||
|
|
||||||
// String returns an opaque representation without conversation content.
|
// String returns an opaque representation without conversation content.
|
||||||
|
|||||||
@@ -51,10 +51,42 @@ func (e *Exchange) Messages() []RenderedMessage {
|
|||||||
if e == nil {
|
if e == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
messages := make([]RenderedMessage, e.count)
|
return e.MessagesWithSuffix()
|
||||||
for current := e; current != nil; current = current.parent {
|
}
|
||||||
start := current.count - len(current.messages)
|
|
||||||
copyRenderedMessages(messages[start:current.count], current.messages)
|
// MessagesWithSuffix returns independently owned history followed by each part.
|
||||||
|
// A nil exchange supplies no history.
|
||||||
|
func (e *Exchange) MessagesWithSuffix(parts ...[]RenderedMessage) []RenderedMessage {
|
||||||
|
count := e.MessageCount()
|
||||||
|
for _, part := range parts {
|
||||||
|
count += len(part)
|
||||||
|
}
|
||||||
|
messages := make([]RenderedMessage, count)
|
||||||
|
e.visitChunks(func(start int, chunk []RenderedMessage) {
|
||||||
|
copyRenderedMessages(messages[start:], chunk)
|
||||||
|
})
|
||||||
|
offset := e.MessageCount()
|
||||||
|
for _, part := range parts {
|
||||||
|
copyRenderedMessages(messages[offset:], part)
|
||||||
|
offset += len(part)
|
||||||
}
|
}
|
||||||
return messages
|
return messages
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// VisitMessages visits indexed history entries without materializing a slice.
|
||||||
|
// Entries and their cache metadata are borrowed read-only values. Visits need
|
||||||
|
// not be in index order; callers can fill a destination using the supplied index.
|
||||||
|
func (e *Exchange) VisitMessages(visit func(int, RenderedMessage)) {
|
||||||
|
e.visitChunks(func(start int, chunk []RenderedMessage) {
|
||||||
|
for index, message := range chunk {
|
||||||
|
visit(start+index, message)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *Exchange) visitChunks(visit func(int, []RenderedMessage)) {
|
||||||
|
for current := e; current != nil; current = current.parent {
|
||||||
|
start := current.count - len(current.messages)
|
||||||
|
visit(start, current.messages)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -51,3 +51,25 @@ func TestExchangeManyTurns(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestExchangeMessagesWithSuffix(t *testing.T) {
|
||||||
|
for _, history := range []*Exchange{nil, CaptureExchange(nil, "source", nil, "answer")} {
|
||||||
|
control := &CacheControl{Type: CacheControlEphemeral, TTL: "1h"}
|
||||||
|
first := []RenderedMessage{{Role: RoleUser, Content: "new", CacheControl: control}}
|
||||||
|
last := []RenderedMessage{{Role: RoleDeveloper, Content: "suffix"}}
|
||||||
|
got := history.MessagesWithSuffix(first, nil, last)
|
||||||
|
want := append(history.Messages(), first...)
|
||||||
|
want = append(want, last...)
|
||||||
|
if !reflect.DeepEqual(got, want) {
|
||||||
|
t.Fatalf("messages = %#v, want %#v", got, want)
|
||||||
|
}
|
||||||
|
got[history.MessageCount()].CacheControl.TTL = "changed"
|
||||||
|
got[history.MessageCount()].Content = "changed"
|
||||||
|
if first[0].Content != "new" || control.TTL != "1h" {
|
||||||
|
t.Fatal("assembled messages alias the caller suffix")
|
||||||
|
}
|
||||||
|
if history != nil && history.Messages()[0].Content != "answer" {
|
||||||
|
t.Fatal("assembly changed retained history")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -482,10 +482,7 @@ func (r *Runner) completePreparationWithStructuredOutput(
|
|||||||
if state.directSessionID != "" {
|
if state.directSessionID != "" {
|
||||||
effectivePrompt.SessionID = state.directSessionID
|
effectivePrompt.SessionID = state.directSessionID
|
||||||
}
|
}
|
||||||
effectivePrompt.Messages = domain.ConcatRenderedMessages(renderedPrompt.Messages, req.AppendedMessages)
|
effectivePrompt.Messages = req.ContinueFrom.MessagesWithSuffix(renderedPrompt.Messages, req.AppendedMessages)
|
||||||
if req.ContinueFrom != nil {
|
|
||||||
effectivePrompt.Messages = domain.ConcatRenderedMessages(req.ContinueFrom.Messages(), effectivePrompt.Messages)
|
|
||||||
}
|
|
||||||
|
|
||||||
end := time.Now().UTC()
|
end := time.Now().UTC()
|
||||||
effectiveModel := state.effectiveModel
|
effectiveModel := state.effectiveModel
|
||||||
|
|||||||
Reference in New Issue
Block a user