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") + } +}