mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-07-22 07:31:05 +08:00
Remove redundant PatchDSL and consolidate component-param injection into override_params (#17079)
## 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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user