From 31cc0069fc4e302c91501b6014264baf705feece Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 14 Nov 2026 01:31:22 +0000 Subject: [PATCH] Demonstrate coordinated conversation workflows --- docs/consumers/pkg-promptkit.md | 106 +++++++++++++++++++++++ docs/roadmap/implementation.md | 2 + examples/go-library/observe/main.go | 83 ++++++++++++------ examples/go-library/observe/main_test.go | 18 ++-- 4 files changed, 175 insertions(+), 34 deletions(-) diff --git a/docs/consumers/pkg-promptkit.md b/docs/consumers/pkg-promptkit.md index f6c53d4..b3eb218 100644 --- a/docs/consumers/pkg-promptkit.md +++ b/docs/consumers/pkg-promptkit.md @@ -695,6 +695,112 @@ application owns any semantic retry budget. When a pre-execution equality check is required, use `PrepareExecution` and compare its opaque rendered-prompt hash before invoking `RunPrepared`. +### Capture And Continue A Conversation + +Capture records the messages actually executed and the final assistant content. +An exchange is an immutable in-process capability; `Details()` returns detached, +encodable history that may contain sensitive content. Ordinary result encoding +does not include the exchange. Continuations imply capture and may branch or use +a different engine, profile, output contract, credential, or session: + +```go +initial, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "proposal", CaptureExchange: true}) +if err != nil { return err } +left, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "refine", ContinueFrom: initial.Exchange}) +if err != nil { return err } +right, err := anotherEngine.Run(ctx, promptkit.RunRequest{PromptID: "alternative", ContinueFrom: initial.Exchange}) +if err != nil { return err } +_, _ = left, right // Independent branches; each result has its own exchange. +``` + +For a frozen execution, set `CaptureExchange` in the request passed to +`PrepareExecution`, then obtain the exchange from `RunPrepared`'s result. +Preparation and validation alone do not produce an exchange. Captured history +precedes newly rendered definition messages and appended messages. A completed +structurally invalid result still has an exchange; an operational error returns +no result. The caller chooses what constitutes acceptance. + +```go +prepared, err := engine.PrepareExecution(ctx, promptkit.RunRequest{PromptID: "proposal", CaptureExchange: true}) +if err != nil { return err } +result, err := engine.RunPrepared(ctx, prepared) +if err != nil { return err } +details := result.Exchange.Details() +_ = details // Explicitly encode or inspect when appropriate. +``` + +### Coordinate Application-Directed Turns + +The following sketches use illustrative application functions. The first records +each newly delivered response before selecting another question: + +```go +outcome, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{ + Start: initial.Exchange, AdditionalTurns: 3, + Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + if turn.Result != nil { + if !turn.Result.Validation.IsValid { return nil, fmt.Errorf("invalid response") } + analysis.Record(turn.Result) + } + question, ok := analysis.NextQuestion() + if !ok { return nil, nil } + return &promptkit.RunRequest{PromptID: "research/investigate", Vars: map[string]string{"question": question}}, nil + }, +}) +``` + +An application can instead start with existing findings, assess each new +candidate, and keep its decisions in a closure: + +```go +outcome, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{ + Start: proposal.Exchange, AdditionalTurns: 3, Observer: branchGroup.Observe, + Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + if turn.Result != nil { + if !turn.Result.Validation.IsValid { return nil, fmt.Errorf("invalid revision") } + candidate, withdrawn, err := decodeRevision(turn.Result.RawOutput) + if err != nil { return nil, err } + if withdrawn { branch.MarkWithdrawn(); return nil, nil } + findings, err := assess(ctx, engine, candidate, validatorGroup.Observe) + if err != nil { return nil, err } + branch.Update(candidate, findings) + if branch.Done() { return nil, nil } + } + return &promptkit.RunRequest{PromptID: branch.FollowUpPrompt(), Vars: branch.Variables()}, nil + }, +}) +``` + +`assess` is application code and explicitly selects its observer on any nested +run. A completed turn releases its admission lease before the callback, so that +assessment may use the same backend. `stopped` and `exhausted` describe control +flow, not content acceptance. Cancellation before callback delivery can return one +completed response in `Undelivered`; errors retain their ordinary identities. + +### Collect Selected Execution Facts + +Groups are independent of continuation and enrollment is explicit: + +```go +var group, validatorGroup promptkit.ExecutionGroup +request.Observer = promptkit.ComposeObservers(diagnosticObserver, group.Observe) +prepared, err := engine.PrepareExecution(ctx, request) // no event yet +if err != nil { return err } +result, err := engine.RunPrepared(ctx, prepared) +if err != nil { return err } +_ = result +// Set CoordinateRequest.Observer to group.Observe for its continuation runs. +// A nested validator selects validatorGroup.Observe on its own request. +summary, validatorSummary := group.Snapshot(), validatorGroup.Snapshot() +_, _ = summary, validatorSummary +``` + +Only the first terminal event per run contributes counts and usage; coverage of +currently observed members is separate from token-usage completeness. Empty +groups report complete, known zero totals, while members without terminals do +not invent totals. Overlapping groups are not generally additive. See the +[offline example](../../examples/go-library/observe/main.go) for complete setup. + ## Inputs, Profiles, And Overrides Use `File`, `Inline`, or `InlineWithURI` to construct request inputs. A request diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 908550a..55c9516 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -585,6 +585,8 @@ retry/repair matrices. ## Stage 12 — Demonstrate The Consumer Workflows +**Status:** Complete. + **Scope:** practical API use and one maintained offline example. 1. Extend `examples/go-library/observe` and its semantic assertions to demonstrate diff --git a/examples/go-library/observe/main.go b/examples/go-library/observe/main.go index 7362cbc..ab2575d 100644 --- a/examples/go-library/observe/main.go +++ b/examples/go-library/observe/main.go @@ -69,12 +69,16 @@ func (l *eventLog) terminal() (promptkit.ExecutionEvent, error) { } type summary struct { - ProposalAttempts uint64 `json:"proposal_attempts"` - ProposalTransientRetries uint64 `json:"proposal_transient_retries"` - ProposalUsageComplete bool `json:"proposal_usage_complete"` - AssessmentAttempts uint64 `json:"assessment_attempts"` - AssessmentTransientRetries uint64 `json:"assessment_transient_retries"` - AssessmentOutcome string `json:"assessment_outcome"` + ProposalAttempts uint64 `json:"proposal_attempts"` + ProposalTransientRetries uint64 `json:"proposal_transient_retries"` + ProposalUsageComplete bool `json:"proposal_usage_complete"` + AssessmentAttempts uint64 `json:"assessment_attempts"` + AssessmentTransientRetries uint64 `json:"assessment_transient_retries"` + AssessmentOutcome string `json:"assessment_outcome"` + CoordinationOutcome promptkit.CoordinationOutcome `json:"coordination_outcome"` + ExhaustedOutcome promptkit.CoordinationOutcome `json:"exhausted_outcome"` + GroupMembers int `json:"group_members"` + GroupAttempts uint64 `json:"group_attempts"` } func run() (summary, error) { @@ -104,8 +108,9 @@ func run() (summary, error) { } var proposalLog eventLog + var executions promptkit.ExecutionGroup proposal, err := engine.Run(context.Background(), promptkit.RunRequest{ - PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Reduce queue cleanup toil.")}, Observer: proposalLog.record, + PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Reduce queue cleanup toil.")}, CaptureExchange: true, Observer: promptkit.ComposeObservers(proposalLog.record, executions.Observe), }) if err != nil { return summary{}, err @@ -118,35 +123,55 @@ func run() (summary, error) { return summary{}, err } var assessmentLog eventLog - assessment, err := engine.Run(context.Background(), promptkit.RunRequest{ - PromptID: "example.assessment", Inputs: map[string]promptkit.ArtifactRef{"proposal": promptkit.Inline(proposal.RawOutput)}, Observer: assessmentLog.record, + decision := "rejected" + coordination, err := engine.Coordinate(context.Background(), promptkit.CoordinateRequest{ + Start: proposal.Exchange, AdditionalTurns: 1, Observer: executions.Observe, + Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + if turn.Result == nil { + return &promptkit.RunRequest{PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Refine the proposal.")}}, nil + } + if !turn.Result.Validation.IsValid { + return nil, fmt.Errorf("follow-up failed structural validation") + } + assessment, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "example.assessment", Inputs: map[string]promptkit.ArtifactRef{"proposal": promptkit.Inline(turn.Result.RawOutput)}, Observer: promptkit.ComposeObservers(assessmentLog.record, executions.Observe)}) + if err != nil { + return nil, err + } + if !assessment.Validation.IsValid { + return nil, fmt.Errorf("assessment failed structural validation") + } + var proposalValue struct { + Risk string `json:"risk"` + } + var assessmentValue struct { + Recommendation string `json:"recommendation"` + } + if err := json.Unmarshal([]byte(turn.Result.RawOutput), &proposalValue); err != nil { + return nil, err + } + if err := json.Unmarshal([]byte(assessment.RawOutput), &assessmentValue); err != nil { + return nil, err + } + if proposalValue.Risk == "low" && assessmentValue.Recommendation == "accept" { + decision = "accepted" + } + return nil, nil + }, }) if err != nil { return summary{}, err } - if !assessment.Validation.IsValid { - return summary{}, fmt.Errorf("assessment failed structural validation") + exhausted, err := engine.Coordinate(context.Background(), promptkit.CoordinateRequest{Start: proposal.Exchange, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + return &promptkit.RunRequest{PromptID: "example.proposal"}, nil + }}) + if err != nil { + return summary{}, err } assessmentTerminal, err := assessmentLog.terminal() if err != nil { return summary{}, err } - var proposalValue struct { - Risk string `json:"risk"` - } - var assessmentValue struct { - Recommendation string `json:"recommendation"` - } - if err := json.Unmarshal([]byte(proposal.RawOutput), &proposalValue); err != nil { - return summary{}, err - } - if err := json.Unmarshal([]byte(assessment.RawOutput), &assessmentValue); err != nil { - return summary{}, err - } - decision := "rejected" - if proposalValue.Risk == "low" && assessmentValue.Recommendation == "accept" { - decision = "accepted" - } + group := executions.Snapshot() return summary{ ProposalAttempts: proposalTerminal.Attempts, ProposalTransientRetries: proposalTerminal.TransientRetries, @@ -154,6 +179,10 @@ func run() (summary, error) { AssessmentAttempts: assessmentTerminal.Attempts, AssessmentTransientRetries: assessmentTerminal.TransientRetries, AssessmentOutcome: decision, + CoordinationOutcome: coordination.Outcome, + ExhaustedOutcome: exhausted.Outcome, + GroupMembers: len(group.Members), + GroupAttempts: group.Attempts, }, nil } diff --git a/examples/go-library/observe/main_test.go b/examples/go-library/observe/main_test.go index c3a3ed9..7f331a1 100644 --- a/examples/go-library/observe/main_test.go +++ b/examples/go-library/observe/main_test.go @@ -12,19 +12,23 @@ func TestObserveExample(t *testing.T) { t.Fatal(err) } var got struct { - ProposalAttempts *int `json:"proposal_attempts"` - ProposalRetries *int `json:"proposal_transient_retries"` - UsageComplete *bool `json:"proposal_usage_complete"` - AssessmentAttempts *int `json:"assessment_attempts"` - AssessmentRetries *int `json:"assessment_transient_retries"` - Outcome string `json:"assessment_outcome"` + ProposalAttempts *int `json:"proposal_attempts"` + ProposalRetries *int `json:"proposal_transient_retries"` + UsageComplete *bool `json:"proposal_usage_complete"` + AssessmentAttempts *int `json:"assessment_attempts"` + AssessmentRetries *int `json:"assessment_transient_retries"` + Outcome string `json:"assessment_outcome"` + CoordinationOutcome string `json:"coordination_outcome"` + ExhaustedOutcome string `json:"exhausted_outcome"` + GroupMembers int `json:"group_members"` + GroupAttempts int `json:"group_attempts"` } if err := json.Unmarshal(output.Bytes(), &got); err != nil { t.Fatal(err) } if got.ProposalAttempts == nil || *got.ProposalAttempts != 2 || got.ProposalRetries == nil || *got.ProposalRetries != 1 || got.UsageComplete == nil || *got.UsageComplete || got.AssessmentAttempts == nil || *got.AssessmentAttempts != 1 || - got.AssessmentRetries == nil || *got.AssessmentRetries != 0 || got.Outcome != "accepted" { + got.AssessmentRetries == nil || *got.AssessmentRetries != 0 || got.Outcome != "accepted" || got.CoordinationOutcome != "stopped" || got.ExhaustedOutcome != "exhausted" || got.GroupMembers != 3 || got.GroupAttempts != 4 { t.Fatalf("unexpected observation summary: %+v", got) } }