diff --git a/internal/ingestion/pipeline/pipeline.go b/internal/ingestion/pipeline/pipeline.go index 2878f29701..e1c1d7acbf 100644 --- a/internal/ingestion/pipeline/pipeline.go +++ b/internal/ingestion/pipeline/pipeline.go @@ -486,56 +486,3 @@ func (p *Pipeline) componentProgressCallback() runtime.ProgressCallback { }) } } - -func PatchDSL(raw string, parserConfig map[string]interface{}) (string, error) { - if len(parserConfig) == 0 { - return raw, nil - } - - var dslMap map[string]any - if err := json.Unmarshal([]byte(raw), &dslMap); err != nil { - return raw, fmt.Errorf("parse dsl: %w", err) - } - - components, _ := dslMap["components"].(map[string]any) - if components == nil { - if inner, ok := dslMap["dsl"].(map[string]any); ok { - components, _ = inner["components"].(map[string]any) - } - if components == nil { - return raw, nil - } - } - for cid, extraVal := range parserConfig { - compVal, ok := components[cid] - if !ok { - continue - } - extraParams, _ := extraVal.(map[string]any) - if extraParams == nil { - continue - } - compMap, _ := compVal.(map[string]any) - if compMap == nil { - continue - } - obj, _ := compMap["obj"].(map[string]any) - if obj == nil { - continue - } - existing, _ := obj["params"].(map[string]any) - if existing == nil { - obj["params"] = extraParams - } else { - for k, v := range extraParams { - existing[k] = v - } - } - } - - patched, err := json.Marshal(dslMap) - if err != nil { - return raw, fmt.Errorf("marshal patched dsl: %w", err) - } - return string(patched), nil -} diff --git a/internal/ingestion/task/pipeline_executor.go b/internal/ingestion/task/pipeline_executor.go index ca7639fff7..2598e8a266 100644 --- a/internal/ingestion/task/pipeline_executor.go +++ b/internal/ingestion/task/pipeline_executor.go @@ -286,8 +286,8 @@ func (s *PipelineExecutor) loadDSLFromCanvas(ctx context.Context, canvasID strin // warnUnknownComponentParams logs a warning for any component id in the // parserConfig whose id is absent from the pipeline DSL. The runtime merge -// (component params -> PatchDSL / override_params) silently drops such -// entries, so we surface them here for operability. API-side validation +// (component params -> override_params) silently drops such entries, so we +// surface them here for operability. API-side validation // already rejects unknown ids on write; this is purely a defensive guard // for legacy/stale rows. func warnUnknownComponentParams(dsl string, parserConfig map[string]any) { @@ -320,24 +320,20 @@ func (s *PipelineExecutor) runPipelineWithDSL(ctx context.Context, dsl string) ( common.InjectExtractorLLMID(parserConfig, s.taskCtx.Tenant.LLMID) // Surface component params whose cpnID is absent from the DSL. The - // runtime merge (PatchDSL / override_params) silently drops such entries; + // runtime merge (override_params) silently drops such entries; // API-side validation already rejects unknown ids on write, so this is a // defensive guard for legacy/stale rows. warnUnknownComponentParams(dsl, parserConfig) - patchedDSL, err := pipelinepkg.PatchDSL(dsl, parserConfig) - if err != nil { - return nil, dsl, fmt.Errorf("patch dsl with parser_config: %w", err) - } pipelineID := "pipeline_" + s.taskCtx.Doc.ID if s.taskCtx.IngestionTask != nil && s.taskCtx.IngestionTask.ID != "" { pipelineID = s.taskCtx.IngestionTask.ID } - pipe, err := pipelinepkg.NewPipelineFromDSL([]byte(patchedDSL), pipelineID, + pipe, err := pipelinepkg.NewPipelineFromDSL([]byte(dsl), pipelineID, pipelinepkg.WithProgressSink(s.progressSink), pipelinepkg.WithDocumentID(s.taskCtx.Doc.ID)) if err != nil { - return nil, patchedDSL, fmt.Errorf("compile pipeline dsl: %w", err) + return nil, dsl, fmt.Errorf("compile pipeline dsl: %w", err) } inputs := map[string]any{} if s.taskCtx.Doc.ID != "" { @@ -349,13 +345,17 @@ func (s *PipelineExecutor) runPipelineWithDSL(ctx context.Context, dsl string) ( inputs["tenant_id"] = s.taskCtx.Tenant.ID inputs["kb_id"] = s.taskCtx.KB.ID - output, err := pipe.Run(ctx, inputs, map[string]interface{}(s.taskCtx.Doc.ParserConfig)) + // Component params from Doc.ParserConfig — including the tenant LLM id + // injected into Extractor components above — are passed to Run as + // override_params, keyed by cpnID with override-wins. The DSL itself is + // compiled unchanged. + output, err := pipe.Run(ctx, inputs, parserConfig) if err != nil { - return nil, patchedDSL, err + return nil, dsl, err } - payload, err := pipelinepkg.ExtractPayload(patchedDSL, output) + payload, err := pipelinepkg.ExtractPayload(dsl, output) if err != nil { - return nil, patchedDSL, err + return nil, dsl, err } - return payload, patchedDSL, nil + return payload, dsl, nil }