From c396ae912e9c9cb6f8cf3f3f5c108c25d7513e77 Mon Sep 17 00:00:00 2001 From: Jack Date: Mon, 20 Jul 2026 11:46:07 +0800 Subject: [PATCH] Remove redundant PatchDSL and consolidate component-param injection into override_params (#17079) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary `Pipeline.Run` already accepts an `override_params` map (keyed by `cpnID`, override-wins) that is merged into each component's params at compile time via `canvas.WithOverrideParams`. `PatchDSL` performed a byte-for-byte equivalent merge by baking `ParserConfig` into the DSL text before compilation, so `runPipelineWithDSL` applied the same `Doc.ParserConfig` twice with no effect. ## Change - Delete the now-dead `PatchDSL` function (`internal/ingestion/pipeline/pipeline.go`); it had a single call site. - In `runPipelineWithDSL`, compile the DSL unchanged and pass the extracted component params to `Run` as `override_params`. ## Behavioral note (important) Previously `PatchDSL` was also the channel that delivered the tenant LLM id into Extractor components: it baked `parserConfig` — after `common.InjectExtractorLLMID(parserConfig, Tenant.LLMID)` had injected `llm_id` — into the DSL. The old `Run` override_params re-cast `Doc.ParserConfig` (which intentionally omits `llm_id`). So removing `PatchDSL` while keeping that re-cast would have silently dropped LLM id injection. This change passes the local `parserConfig` (the LLMID-injected copy) to `Run`, preserving the prior behavior. The DSL string returned by `runPipelineWithDSL` is only consumed by `recordPipelineLog` (structure storage) and `ExtractPayload` (which reads the terminal structure, not params), so returning the original DSL is equivalent. ## Test plan - `bash build.sh --test ./internal/ingestion/task/... ./internal/ingestion/pipeline/...` — both packages pass. - `gofmt` clean; pre-commit hooks pass. Co-authored-by: Claude --- internal/ingestion/pipeline/pipeline.go | 53 -------------------- internal/ingestion/task/pipeline_executor.go | 28 +++++------ 2 files changed, 14 insertions(+), 67 deletions(-) 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 }