From aa512d0fe50c7512c76ce106379b7347b9bc0da3 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Thu, 4 Jun 2026 00:33:38 -0300 Subject: [PATCH] feat(ai): structured output for ai/chat & ai/agent via tool-forced respond (ADR-0030) Opt-in: an outputSchema input (list of {name,type,required,description}) makes the node return a validated object instead of free-form text. A single forced "respond" tool carries the schema (ParameterSchemaToJSONSchema) with ToolChoice=required; the tool-call args are parsed and validated, with one bounded repair retry. Works on every provider (all support tool calling). The agent runs its normal loop then coerces the final answer. Absent outputSchema = unchanged. Provider-native response_format (Option B) deferred. ADR-0030 Proposed -> Accepted. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../adr/0030-structured-output-enforcement.md | 13 +- docs/adr/README.md | 9 +- internal/packages/functions/ai/agent.go | 22 ++- internal/packages/functions/ai/chat.go | 26 ++++ internal/packages/functions/ai/structured.go | 147 ++++++++++++++++++ .../packages/functions/ai/structured_test.go | 104 +++++++++++++ 6 files changed, 315 insertions(+), 6 deletions(-) create mode 100644 internal/packages/functions/ai/structured.go create mode 100644 internal/packages/functions/ai/structured_test.go diff --git a/docs/adr/0030-structured-output-enforcement.md b/docs/adr/0030-structured-output-enforcement.md index cc310d3..f870f48 100644 --- a/docs/adr/0030-structured-output-enforcement.md +++ b/docs/adr/0030-structured-output-enforcement.md @@ -1,6 +1,6 @@ # 0030. Structured / JSON output enforcement for `ai/chat` & `ai/agent` -- Status: Proposed +- Status: Accepted (tool-forced "respond" path shipped; provider-native `response_format` deferred) - Date: 2026-06-02 - Deciders: FUSE maintainers @@ -68,6 +68,17 @@ provider supports native structured output, use it (B); otherwise fall back to t ## More Information +- **Shipped (Option C, tool-forced — universal)**: `ai/chat` and `ai/agent` accept an optional + `outputSchema` input (a list of `{name,type,required,description}` ParameterSchema). When set, the + node offers a single forced `respond` tool whose parameters are the JSON Schema + (`ParameterSchemaToJSONSchema`) with `ToolChoice: "required"`, parses the tool-call arguments, and + validates them against the schema; one bounded repair retry on invalid output, then a + `FunctionError`. The validated object becomes the node's `output` (instead of free-form text); the + agent runs its normal loop, then coerces the final answer. Absent `outputSchema` = unchanged. + Works on every provider (all support tool calling). Logic in + `internal/packages/functions/ai/structured.go`. **Deferred (Option B optimization)**: + provider-native `response_format` / `json_schema` with a per-provider capability map for providers + that support it natively. - Reuse: `internal/packages/functions/ai/tools.go` (`ParameterSchemaToJSONSchema`), `pkg/workflow/metadata.go` (`ParameterSchema`). - Related: [ADR-0006](0006-llm-provider-abstraction-and-multi-provider-strategy.md), diff --git a/docs/adr/README.md b/docs/adr/README.md index 836544d..0a29574 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -56,7 +56,7 @@ copying [`template.md`](template.md). | 0027 | [Async tool invocation via a sub-execution channel](0027-async-tool-invocation-sub-execution-channel.md) | Proposed | 2026-06-02 | | 0028 | [Agent prompt/context & conversation-memory model](0028-agent-prompt-context-and-memory-model.md) | Accepted | 2026-06-02 | | 0029 | [LLM cost & token-usage tracking and budgets](0029-llm-cost-and-usage-tracking-and-budgets.md) | Accepted | 2026-06-02 | -| 0030 | [Structured/JSON output enforcement for ai nodes](0030-structured-output-enforcement.md) | Proposed | 2026-06-02 | +| 0030 | [Structured/JSON output enforcement for ai nodes](0030-structured-output-enforcement.md) | Accepted | 2026-06-02 | | 0031 | [Settings, secrets & environments: a SecretStore seam](0031-settings-secrets-and-environments.md) | Accepted | 2026-06-02 | | 0032 | [Sub-workflow composition: child workflows as first-class instances](0032-sub-workflow-composition.md) | Accepted | 2026-06-03 | | 0033 | [Dependency injection & app composition with uber-go/fx](0033-dependency-injection-and-app-composition.md) | Accepted | 2026-06-03 | @@ -66,8 +66,9 @@ copying [`template.md`](template.md). ADRs **0026–0030** are a cohesive **agent-capabilities** series for the next phase of the AI agent ([ADR-0005](0005-ai-agents-as-workflow-nodes-phased-roadmap.md)) — orchestrator mode (0026), async tool invocation (0027), prompt/context & memory (0028), cost/usage tracking & budgets (0029), and -structured-output enforcement (0030). The leaf capabilities are being implemented first: **0029** -Phase A (usage visibility) is `Accepted`/shipped; 0028 and 0030 follow; the larger orchestrator -(0026) + async-tools (0027) pair stays `Proposed` until reassessed. **0025** (browser-automation +structured-output enforcement (0030). The leaf capabilities shipped first: **0028** +(context policy), **0029** Phase A (usage visibility), and **0030** (structured output) are +`Accepted`. The larger orchestrator (0026) + async-tools (0027) pair stays `Proposed` until +reassessed. **0025** (browser-automation package) is an independent, parallel stream, not part of that series. When an ADR is implemented its status moves to `Accepted` and its "More Information" records what shipped (as 0031 does). diff --git a/internal/packages/functions/ai/agent.go b/internal/packages/functions/ai/agent.go index ec78d86..38807a2 100644 --- a/internal/packages/functions/ai/agent.go +++ b/internal/packages/functions/ai/agent.go @@ -45,6 +45,7 @@ func AgentFunctionMetadata() workflow.FunctionMetadata { {Name: "allowedTools", Type: "array", Required: false, Description: "Optional allowlist of full function ids the agent may use as tools; empty means all eligible tools"}, {Name: "maxContextTokens", Type: "int", Required: false, Description: "Optional token budget for the running transcript (approximate); 0/absent disables trimming (ADR-0028)"}, {Name: "contextStrategy", Type: "string", Required: false, Description: "When over maxContextTokens: 'drop-oldest' (default) or 'summarize' (an extra LLM call summarizes dropped turns)"}, + {Name: "outputSchema", Type: "array", Required: false, Description: "Optional list of {name,type,required,description} fields; when set the final output is a validated object matching this schema (ADR-0030)"}, }, Edges: workflow.InputEdgeMetadata{ Count: 0, @@ -90,6 +91,7 @@ func makeAgentFunction(providers llm.Registry, tools ToolRegistry, usage UsageRe usage: usage, maxContextTokens: input.GetInt("maxContextTokens"), contextStrategy: contextStrategyOrDefault(input.GetStr("contextStrategy")), + outputSchema: parseOutputSchema(input), } messages := make([]llm.Message, 0, 4) @@ -136,6 +138,7 @@ type agentExecutor struct { usage UsageRecorder maxContextTokens int contextStrategy string + outputSchema []workflow.ParameterSchema } // run drives the reasoning loop until a final answer, an error, or the iteration @@ -173,6 +176,17 @@ func (e *agentExecutor) run(ctx context.Context, messages []llm.Message) workflo messages = append(messages, resp.Message) if len(resp.Message.ToolCalls) == 0 { + // Structured output (ADR-0030): coerce the final answer into the requested schema. + if len(e.outputSchema) > 0 { + obj, u, serr := structuredOutput(ctx, e.provider, e.model, messages, e.outputSchema, e.usage, AgentFunctionID) + if serr != nil { + return errorOutput(fmt.Sprintf("ai/agent: structured output failed: %v", serr)) + } + totalUsage.PromptTokens += u.PromptTokens + totalUsage.CompletionTokens += u.CompletionTokens + totalUsage.TotalTokens += u.TotalTokens + return successOutputData(obj, totalUsage, steps) + } return successOutput(resp.Message.Content, totalUsage, steps) } @@ -341,8 +355,14 @@ func marshalToolContent(data map[string]any) string { // successOutput builds the agent's final successful output. func successOutput(answer string, usage llm.Usage, steps []map[string]any) workflow.FunctionOutput { + return successOutputData(answer, usage, steps) +} + +// successOutputData builds the agent's final output; output is the answer text or, for structured +// output (ADR-0030), the validated object. +func successOutputData(output any, usage llm.Usage, steps []map[string]any) workflow.FunctionOutput { return workflow.NewFunctionSuccessOutput(map[string]any{ - "output": answer, + "output": output, "usage": map[string]any{ "promptTokens": usage.PromptTokens, "completionTokens": usage.CompletionTokens, diff --git a/internal/packages/functions/ai/chat.go b/internal/packages/functions/ai/chat.go index 8e8eb03..efecfee 100644 --- a/internal/packages/functions/ai/chat.go +++ b/internal/packages/functions/ai/chat.go @@ -57,6 +57,12 @@ func ChatFunctionMetadata() workflow.FunctionMetadata { Required: false, Description: "Sampling temperature; if omitted the provider default is used", }, + { + Name: "outputSchema", + Type: "array", + Required: false, + Description: "Optional list of {name,type,required,description} fields; when set the output is a validated object matching this schema (ADR-0030)", + }, }, Edges: workflow.InputEdgeMetadata{ Count: 0, @@ -107,6 +113,7 @@ func makeChatFunction(providers llm.Registry, usage UsageRecorder) workflow.Func Messages: messages, Temperature: optionalTemperature(input), } + outputSchema := parseOutputSchema(input) // Provider resolution and the completion run in their own goroutine and report back via // Finish so the WorkflowFunc pool worker is freed immediately (mirrors logic/timer). @@ -123,6 +130,25 @@ func makeChatFunction(providers llm.Registry, usage UsageRecorder) workflow.Func return } + // Structured output (ADR-0030): coerce the answer into the requested schema. + if len(outputSchema) > 0 { + obj, u, serr := structuredOutput(ctx, provider, req.Model, messages, outputSchema, usage, ChatFunctionID) + if serr != nil { + log.Error().Err(serr).Str("provider", provider.Name()).Msg("ai/chat structured output failed") + execInfo.Finish(workflow.NewFunctionOutput(workflow.FunctionError, map[string]any{"error": serr.Error()})) + return + } + execInfo.Finish(workflow.NewFunctionSuccessOutput(map[string]any{ + "output": obj, + "usage": map[string]any{ + "promptTokens": u.PromptTokens, + "completionTokens": u.CompletionTokens, + "totalTokens": u.TotalTokens, + }, + })) + return + } + resp, err := provider.Chat(ctx, req) if err != nil { usage.RecordCall(ChatFunctionID, provider.Name(), req.Model, "error") diff --git a/internal/packages/functions/ai/structured.go b/internal/packages/functions/ai/structured.go new file mode 100644 index 0000000..6352b51 --- /dev/null +++ b/internal/packages/functions/ai/structured.go @@ -0,0 +1,147 @@ +package ai + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/open-source-cloud/fuse/pkg/llm" + "github.com/open-source-cloud/fuse/pkg/workflow" +) + +// respondToolName is the synthetic tool the model is forced to call to deliver structured output. +const respondToolName = "respond" + +// structuredRepairAttempts bounds how many times structuredOutput re-asks on invalid output. +const structuredRepairAttempts = 2 + +// parseOutputSchema reads the optional `outputSchema` input (a list of ParameterSchema-shaped +// objects) into typed schema params. Absent/invalid returns nil (free-form output). +func parseOutputSchema(input *workflow.FunctionInput) []workflow.ParameterSchema { + raw := input.GetAnySliceOrDefault("outputSchema", nil) + if len(raw) == 0 { + return nil + } + b, err := json.Marshal(raw) + if err != nil { + return nil + } + var schema []workflow.ParameterSchema + if json.Unmarshal(b, &schema) != nil { + return nil + } + return schema +} + +// structuredOutput coerces a completion into a typed object matching schema (ADR-0030). It offers a +// single forced "respond" tool whose parameters are the JSON Schema of the output, parses the tool +// call's arguments, and validates them; on invalid/malformed output it re-asks once. Tokens are +// recorded via usage. This tool-forced path works on every provider (all support tool calling); +// provider-native response_format is a deferred optimization. +func structuredOutput(ctx context.Context, provider llm.Provider, model string, messages []llm.Message, schema []workflow.ParameterSchema, usage UsageRecorder, function string) (map[string]any, llm.Usage, error) { + respondTool := llm.Tool{ + Name: respondToolName, + Description: "Return the final answer as JSON arguments matching the required schema.", + Parameters: ParameterSchemaToJSONSchema(schema), + } + + convo := messages + var total llm.Usage + var lastErr error + for attempt := 0; attempt < structuredRepairAttempts; attempt++ { + resp, err := provider.Chat(ctx, llm.ChatRequest{ + Model: model, + Messages: convo, + Tools: []llm.Tool{respondTool}, + ToolChoice: "required", + }) + if err != nil { + usage.RecordCall(function, provider.Name(), model, "error") + return nil, total, fmt.Errorf("structured output call failed: %w", err) + } + usage.RecordCall(function, provider.Name(), model, "success") + usage.RecordUsage(function, provider.Name(), model, resp.Usage) + total.PromptTokens += resp.Usage.PromptTokens + total.CompletionTokens += resp.Usage.CompletionTokens + total.TotalTokens += resp.Usage.TotalTokens + + obj, perr := parseRespond(resp.Message) + if perr != nil { + lastErr = perr + } else if verr := validateAgainstSchema(obj, schema); verr != nil { + lastErr = verr + } else { + return obj, total, nil + } + // Repair: re-ask from the ORIGINAL messages plus a note (avoids a dangling tool_call turn). + convo = append(append([]llm.Message{}, messages...), llm.Message{ + Role: llm.RoleUser, + Content: fmt.Sprintf("The previous response was not valid (%v). Call the %q tool again with arguments that match the schema.", lastErr, respondToolName), + }) + } + return nil, total, fmt.Errorf("structured output did not match the schema after %d attempts: %w", structuredRepairAttempts, lastErr) +} + +// parseRespond extracts the structured object from a model turn: preferring the `respond` tool +// call's arguments, falling back to parsing the message content as a JSON object. +func parseRespond(msg llm.Message) (map[string]any, error) { + for _, tc := range msg.ToolCalls { + if tc.Name == respondToolName || len(msg.ToolCalls) == 1 { + var obj map[string]any + if err := json.Unmarshal(tc.Arguments, &obj); err != nil { + return nil, fmt.Errorf("tool arguments are not a JSON object: %w", err) + } + return obj, nil + } + } + if msg.Content != "" { + var obj map[string]any + if err := json.Unmarshal([]byte(msg.Content), &obj); err == nil { + return obj, nil + } + } + return nil, fmt.Errorf("model did not return structured output") +} + +// validateAgainstSchema checks required fields are present and that present fields are of a +// compatible JSON type. It is intentionally lenient (a guard, not a full JSON-Schema validator). +func validateAgainstSchema(obj map[string]any, schema []workflow.ParameterSchema) error { + for _, p := range schema { + v, ok := obj[p.Name] + if !ok { + if p.Required { + return fmt.Errorf("missing required field %q", p.Name) + } + continue + } + if !valueMatchesType(v, p.Type) { + return fmt.Errorf("field %q has wrong type (want %s)", p.Name, p.Type) + } + } + return nil +} + +// valueMatchesType reports whether a JSON-decoded value is compatible with a FUSE schema type. +func valueMatchesType(v any, t string) bool { + switch t { + case "", "any": + return true + case "string": + _, ok := v.(string) + return ok + case "bool": + _, ok := v.(bool) + return ok + case "int", "float": + _, ok := v.(float64) // JSON numbers decode to float64 + return ok + case "array": + _, ok := v.([]any) + return ok + case "map", "object": + _, ok := v.(map[string]any) + return ok + default: + return true + } +} diff --git a/internal/packages/functions/ai/structured_test.go b/internal/packages/functions/ai/structured_test.go new file mode 100644 index 0000000..1d9b6bd --- /dev/null +++ b/internal/packages/functions/ai/structured_test.go @@ -0,0 +1,104 @@ +package ai + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/open-source-cloud/fuse/pkg/llm" + "github.com/open-source-cloud/fuse/pkg/workflow" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var testOutputSchema = []workflow.ParameterSchema{ + {Name: "answer", Type: "string", Required: true}, + {Name: "score", Type: "int", Required: false}, +} + +func TestValidateAgainstSchema(t *testing.T) { + t.Parallel() + assert.NoError(t, validateAgainstSchema(map[string]any{"answer": "hi", "score": float64(5)}, testOutputSchema)) + assert.NoError(t, validateAgainstSchema(map[string]any{"answer": "hi"}, testOutputSchema)) // optional omitted + assert.Error(t, validateAgainstSchema(map[string]any{"score": float64(5)}, testOutputSchema)) // required missing + assert.Error(t, validateAgainstSchema(map[string]any{"answer": 1}, testOutputSchema)) // wrong type +} + +func TestParseRespond(t *testing.T) { + t.Parallel() + + obj, err := parseRespond(llm.Message{ToolCalls: []llm.ToolCall{ + {Name: respondToolName, Arguments: json.RawMessage(`{"answer":"hi"}`)}, + }}) + require.NoError(t, err) + assert.Equal(t, "hi", obj["answer"]) + + obj, err = parseRespond(llm.Message{Content: `{"answer":"yo"}`}) + require.NoError(t, err) + assert.Equal(t, "yo", obj["answer"]) + + _, err = parseRespond(llm.Message{Content: "not json"}) + assert.Error(t, err) +} + +func TestStructuredOutput_ToolForced(t *testing.T) { + t.Parallel() + prov := &stubProvider{ + name: "stub", + resp: llm.ChatResponse{ + Message: llm.Message{ + Role: llm.RoleAssistant, + ToolCalls: []llm.ToolCall{{ID: "1", Name: respondToolName, Arguments: json.RawMessage(`{"answer":"42","score":7}`)}}, + }, + Usage: llm.Usage{PromptTokens: 5, CompletionTokens: 2, TotalTokens: 7}, + }, + } + obj, u, err := structuredOutput(context.Background(), prov, "m", + []llm.Message{{Role: llm.RoleUser, Content: "q"}}, testOutputSchema, NopUsageRecorder{}, ChatFunctionID) + require.NoError(t, err) + assert.Equal(t, "42", obj["answer"]) + assert.Equal(t, float64(7), obj["score"]) + assert.Equal(t, 7, u.TotalTokens) + // The forced request offered exactly the respond tool. + require.Len(t, prov.last.Tools, 1) + assert.Equal(t, respondToolName, prov.last.Tools[0].Name) + assert.Equal(t, "required", prov.last.ToolChoice) +} + +func TestChat_StructuredOutputDeliversObject(t *testing.T) { + prov := &stubProvider{ + name: "stub", + resp: llm.ChatResponse{ + Message: llm.Message{ + Role: llm.RoleAssistant, + ToolCalls: []llm.ToolCall{{ID: "1", Name: respondToolName, Arguments: json.RawMessage(`{"answer":"structured"}`)}}, + }, + }, + } + reg := llm.NewStaticRegistry(map[string]llm.Provider{"stub": prov}, "stub") + + fnInput, err := workflow.NewFunctionInputWith(map[string]any{ + "input": "hello", + "outputSchema": []any{ + map[string]any{"name": "answer", "type": "string", "required": true}, + }, + }) + require.NoError(t, err) + done := make(chan workflow.FunctionOutput, 1) + execInfo := workflow.NewExecutionInfo("wf-1", "exec-1", "", fnInput) + execInfo.Finish = func(out workflow.FunctionOutput) { done <- out } + + res, err := makeChatFunction(reg, NopUsageRecorder{})(execInfo) + require.NoError(t, err) + require.True(t, res.Async) + select { + case out := <-done: + require.Equal(t, workflow.FunctionSuccess, out.Status) + obj, ok := out.Data["output"].(map[string]any) + require.True(t, ok, "output should be a structured object") + assert.Equal(t, "structured", obj["answer"]) + case <-time.After(2 * time.Second): + t.Fatal("timed out") + } +}