zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-08-04 14:50:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Align Go ingestion boundaries with Python (#16647)

Browse Source 
Moves doc_id blob resolution into Parser, tightens chunker/tokenizer to
Python output_format semantics, updates extractor list handling, and
fixes real-template integration tests.
This commit is contained in:
Zhichang Yu
2026-07-05 20:43:52 +08:00
committed by GitHub
parent 0fcfb38365
commit 014c3f634f
119 changed files with 18083 additions and 4712 deletions
Download Patch File Download Diff File Expand all files Collapse all files

514
internal/agent/component/pipeline_chunker.go
View File

@@ -1,514 +0,0 @@
//
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// PipelineChunker component (T3) — partial port of python
// `agent/component/pipeline_chunker.py` (PR #15068).
//
// SCOPE (honest):
//
// - WHITELIST: every parser_id in the python
// `_PARSER_MODULES` table is accepted (general/naive/paper/
// book/presentation/manual/laws/qa/table/resume/picture/one/
// audio/email/tag). Check() rejects anything else.
//
// - DISPATCH: parser_id drives the chunk-engine split
// strategy. The Go chunk engine exposes sentence / paragraph /
// char / paragraph splits; we map the python parser_ids to
// those strategies (see parserToSplitStrategy). Per-parser
// knobs (paper's double-column handling, table's HTML
// reconstruction, laws' article-aware boundaries, …) are
// not ported yet — a canvas author who picks `paper` gets a
// paragraph split, not the python paper parser's column-aware
// behaviour.
//
// - TEXT INPUT: "text" / "content" / "file_bytes" (interpreted
// as UTF-8 text) flow through the chunk engine. This matches
// the python "text file" path.
//
// "file_ref" is the bytes-container form used by the other Go
// agent components (ExcelProcessor, etc.). It accepts
// []byte OR a base64-encoded string. Raw text MUST go in
// "text" / "content" — a string file_ref that is not valid
// base64 is rejected with a clear error so a caller that
// mistakenly hands plain text doesn't see it silently
// reinterpreted (a "try base64, fall back" policy would
// rewrite any plain text that happens to satisfy the
// base64 alphabet, e.g. "Zm9v" → "foo"). The contract is
// therefore: file_ref is always raw bytes (or base64 of
// them); text/content is always UTF-8 text.
//
// - FILE-FORMAT EXTRACTION (PDF, DOCX, XLSX, PPT, images, …):
// NOT PORTED. The python side uses `rag.app.<parser>.chunk
// (filename)` which extracts text from the file format
// AND chunks in one step. The Go side does not yet have a
// unified `rag.app.*.chunk` dispatch — `deepdoc/parser` is
// still Python-only. A caller that supplies binary
// (non-UTF-8) bytes is rejected with an explicit error so
// the canvas author knows to add text extraction upstream.
// This is a follow-up that lands with the deepdoc/parser
// port.
//
// - NO EMBEDDING / NO PERSISTENCE: chunks live only in
// canvas variables for the run, exactly as in the python
// fix.
package component
import (
"context"
"encoding/base64"
"fmt"
"unicode/utf8"
"ragflow/internal/agent/runtime"
"ragflow/internal/ingestion"
"ragflow/internal/ingestion/chunk"
)
const componentNamePipelineChunker = "PipelineChunker"
// supportedParserIDs mirrors the python `_PARSER_MODULES` keys.
// The whitelist is enforced at component-construction time
// so a misspelled parser_id is caught at canvas-build time,
// not at run time.
var supportedPipelineParserIDs = map[string]struct{}{
"general": {},
"naive": {},
"paper": {},
"book": {},
"presentation": {},
"manual": {},
"laws": {},
"qa": {},
"table": {},
"resume": {},
"picture": {},
"one": {},
"audio": {},
"email": {},
"tag": {},
}
// pipelineChunkerParam is the static DSL configuration for a
// PipelineChunker node. Fields mirror python
// PipelineChunkerParam.__init__ defaults.
type pipelineChunkerParam struct {
ParserID string `json:"parser_id"` // whitelisted; drives split strategy
Lang string `json:"lang"` // python-only knob, surfaced for DSL parity
FromPage int `json:"from_page"` // python-only knob, surfaced for DSL parity
ToPage int `json:"to_page"` // python-only knob, surfaced for DSL parity
ParserConfig map[string]any `json:"parser_config"` // python-only knob, surfaced for DSL parity
}
// Update copies a fresh param map into the receiver.
func (p *pipelineChunkerParam) Update(conf map[string]any) error {
if conf == nil {
conf = map[string]any{}
}
p.ParserID, _ = conf["parser_id"].(string)
if p.ParserID == "" {
p.ParserID = "naive"
}
p.Lang, _ = conf["lang"].(string)
if p.Lang == "" {
p.Lang = "English"
}
if v, ok := conf["from_page"].(float64); ok {
p.FromPage = int(v)
} else if v, ok := conf["from_page"].(int); ok {
p.FromPage = v
}
if v, ok := conf["to_page"].(float64); ok {
p.ToPage = int(v)
} else if v, ok := conf["to_page"].(int); ok {
p.ToPage = v
}
if cfg, ok := conf["parser_config"].(map[string]any); ok {
p.ParserConfig = cfg
} else {
p.ParserConfig = map[string]any{}
}
return nil
}
// Check validates the parser_id whitelist, page range, and
// parser_config shape. Mirrors python
// PipelineChunkerParam.check().
func (p *pipelineChunkerParam) Check() error {
if _, ok := supportedPipelineParserIDs[p.ParserID]; !ok {
return fmt.Errorf("PipelineChunker: parser_id %q is not supported (allowed: %v)",
p.ParserID, pipelineChunkerWhitelistOrdered())
}
if p.FromPage < 0 {
return fmt.Errorf("PipelineChunker: from_page must be non-negative (got %d)", p.FromPage)
}
if p.ToPage < 0 {
return fmt.Errorf("PipelineChunker: to_page must be non-negative (got %d)", p.ToPage)
}
if p.FromPage > p.ToPage {
return fmt.Errorf("PipelineChunker: from_page (%d) must be <= to_page (%d)",
p.FromPage, p.ToPage)
}
if p.ParserConfig == nil {
return fmt.Errorf("PipelineChunker: parser_config must be a dict")
}
return nil
}
func pipelineChunkerWhitelistOrdered() []string {
out := make([]string, 0, len(supportedPipelineParserIDs))
for k := range supportedPipelineParserIDs {
out = append(out, k)
}
for i := 1; i < len(out); i++ {
for j := i; j > 0 && out[j-1] > out[j]; j-- {
out[j-1], out[j] = out[j], out[j-1]
}
}
return out
}
// parserToSplitStrategy maps a python parser_id to the Go
// chunk engine's split strategy. The Go chunk engine exposes
// sentence / paragraph / char / paragraph splits — these
// correspond loosely to the python "naive" / "paper" /
// "table" strategies but do not implement the parser-specific
// extraction (PDF column detection, table HTML reconstruction,
// etc.). The mapping below is the conservative best-effort
// port: pick the split granularity that most matches the
// python parser's intent. A canvas author who needs the full
// parser-specific behaviour must wait for the deepdoc/parser
// port to land.
func parserToSplitStrategy(parserID string) string {
switch parserID {
case "general", "naive", "book", "presentation",
"manual", "qa", "resume", "email", "tag":
return "paragraph"
case "paper", "laws":
// python paper/laws chunk on sentence boundaries
// within article/column scopes; the Go split engine
// does not have a "scoped sentence" split, so we
// fall back to sentence-level. The structural loss
// is documented in the package comment.
return "sentence"
case "table":
// python table chunker emits one chunk per row.
// The Go side has no row-aware split; char-split
// at a 1024 rune size approximates it. The
// mismatch is documented.
return "char"
case "picture", "one", "audio":
// picture/one/audio produce a single chunk per
// file. The Go side has no "single-chunk" split;
// paragraph split on a single-paragraph input
// collapses to one chunk.
return "paragraph"
default:
return "paragraph"
}
}
// PipelineChunkerComponent runs the configured chunker against
// the input text and returns the chunks as plain text (no
// embedding, no persistence) for downstream Agent nodes.
//
// Output shape:
//
// chunks — []string of plain-text chunks
// chunks_full — []map[string]any with at minimum {text, ...}
// summary — short human-readable summary (parser_id + chunk count)
type PipelineChunkerComponent struct {
name string
param pipelineChunkerParam
}
// NewPipelineChunkerComponent constructs a PipelineChunker from
// the DSL param map. Errors here surface as canvas compile
// failures so a bad parser_id is caught at canvas-build time
// rather than mid-run.
func NewPipelineChunkerComponent(params map[string]any) (Component, error) {
p := &pipelineChunkerParam{}
if err := p.Update(params); err != nil {
return nil, fmt.Errorf("PipelineChunker: param update: %w", err)
}
if err := p.Check(); err != nil {
return nil, fmt.Errorf("PipelineChunker: param check: %w", err)
}
return &PipelineChunkerComponent{
name: componentNamePipelineChunker,
param: *p,
}, nil
}
// Name returns the registered component name.
func (c *PipelineChunkerComponent) Name() string { return c.name }
// Stream is a synchronous facade over Invoke.
func (c *PipelineChunkerComponent) Stream(ctx context.Context, inputs map[string]any) (<-chan map[string]any, error) {
out, err := c.Invoke(ctx, inputs)
if err != nil {
return nil, err
}
ch := make(chan map[string]any, 1)
ch <- out
close(ch)
return ch, nil
}
// Inputs returns the parameter metadata. The component reads
// any of the following from the inputs map, in order:
//
// text (string) — raw UTF-8 text (primary)
// content (string) — alias for "text"
// file_ref ([]byte | base64 str) — file bytes container.
// Mirrors the ExcelProcessor
// contract. The []byte form
// is the in-process caller's
// normal form; the string
// form is HTTP / JSON
// callers' normal form
// (base64). Raw text MUST
// go in "text" / "content".
// file_bytes ([]byte | base64 str) — alias for "file_ref"
// under a more honest
// name. Same contract.
//
// Binary file bytes must be text-extracted upstream until the
// deepdoc/parser port lands; non-UTF-8 bytes are rejected
// with a clear "not yet ported" error.
func (c *PipelineChunkerComponent) Inputs() map[string]string {
return map[string]string{
"text": "Plain-text input. The chunker slices this into downstream chunks.",
"content": "Alias for \"text\".",
"file_ref": "File bytes ([]byte) or base64-encoded string. NOT a state ref name. Raw text goes in \"text\" / \"content\".",
"file_bytes": "Alias for \"file_ref\" ([]byte or base64-encoded string). Same encoding contract.",
}
}
// Outputs returns the public surface that downstream Agent
// nodes can wire into.
func (c *PipelineChunkerComponent) Outputs() map[string]string {
return map[string]string{
"chunks": "list[string]: plain-text chunks.",
"chunks_full": "list[object]: per-chunk metadata (text + size + index).",
"summary": "string: short human-readable summary.",
}
}
// Invoke runs the chunker against the input.
//
// Inputs contract:
//
// "text" — (preferred) the already-extracted plain text
// "content" — alias for "text"
// "file_bytes" — raw bytes, MUST be valid UTF-8 (text formats
// only; binary formats return an explicit
// "deepdoc/parser not yet ported" error)
//
// Empty input returns the no-chunks sentinel.
//
// Non-UTF-8 bytes are rejected with an explicit "file-format
// extraction not ported" error so the canvas author is told
// to add text extraction upstream (or use the python canvas).
func (c *PipelineChunkerComponent) Invoke(ctx context.Context, inputs map[string]any) (map[string]any, error) {
if _, _, err := runtime.GetStateFromContext[*runtime.CanvasState](ctx); err != nil {
return nil, fmt.Errorf("PipelineChunker: %w", err)
}
text, err := readPipelineInputText(inputs)
if err != nil {
return nil, err
}
if text == "" {
return map[string]any{
"chunks": []string{},
"chunks_full": []map[string]any{},
"summary": "no input text",
}, nil
}
strategy := parserToSplitStrategy(c.param.ParserID)
dsl := buildPipelineChunkerDSL(strategy)
plan, err := ingestion.NewChunkEngine().Compile(dsl)
if err != nil {
return nil, fmt.Errorf("PipelineChunker: compile (strategy=%s): %w", strategy, err)
}
result, err := ingestion.NewChunkEngine().Execute(plan, text)
if err != nil {
return nil, fmt.Errorf("PipelineChunker: execute: %w", err)
}
return chunkerOutputs(result, c.param.ParserID), nil
}
// readPipelineInputText returns the input text from any of the
// supported keys. Non-UTF-8 binary inputs are rejected with an
// explicit "extraction not ported" error so the canvas author
// gets a clear signal instead of a silent garbled chunk.
//
// Supported keys (first match wins):
//
// text (string) — raw UTF-8 text (primary)
// content (string) — alias for "text"
// file_ref ([]byte | base64 str) — file bytes container.
// Mirrors the ExcelProcessor
// contract. The []byte form
// is the in-process caller's
// normal form; the string
// form is HTTP / JSON callers'
// normal form (base64). Raw
// text MUST go in "text" /
// "content" — see
// decodeFileRefString for
// the rationale.
// file_bytes ([]byte | base64 str) — alias for file_ref under a
// more honest name. Same
// encoding contract.
func readPipelineInputText(inputs map[string]any) (string, error) {
if inputs == nil {
return "", nil
}
if v, ok := inputs["text"].(string); ok {
return v, nil
}
if v, ok := inputs["content"].(string); ok {
return v, nil
}
// file_ref accepts []byte or a base64-encoded string,
// matching the ExcelProcessor contract. The orchestrator
// is responsible for any state-ref → bytes resolution.
if b, ok := inputs["file_ref"].([]byte); ok {
return validateAndDecodeBytes(b)
}
if s, ok := inputs["file_ref"].(string); ok && s != "" {
return decodeFileRefString(s)
}
// file_bytes is the same bytes contract under a more
// honest name; both keys map to the same handler.
if b, ok := inputs["file_bytes"].([]byte); ok {
return validateAndDecodeBytes(b)
}
// file_bytes also accepts a base64-encoded string, matching
// file_ref's contract — JSON callers hand the orchestrator a
// base64 blob under whichever key the upstream component
// happens to emit, so we accept both forms rather than
// silently dropping a perfectly-valid payload that used
// "file_bytes" instead of "file_ref". Same strict-base64 rule
// as decodeFileRefString: no fall-through to raw text.
if s, ok := inputs["file_bytes"].(string); ok && s != "" {
return decodeFileRefString(s)
}
return "", nil
}
// decodeFileRefString treats a file_ref string as STRICTLY
// base64-encoded raw bytes. There is no "fall back to raw
// text" path: a "try base64, fall back" policy would silently
// rewrite any plain-text input that happens to satisfy the
// base64 alphabet (e.g. "Zm9v" → "foo", "Q29kZUNvbnZlcnQ"
// → "CodeConvert"). The contract is unambiguous: file_ref
// string = base64 of the raw bytes. Callers that have raw
// text should use the "text" / "content" keys.
//
// The decoded bytes still flow through validateAndDecodeBytes
// so a PDF / DOCX with no upstream extraction surfaces a
// loud "not yet ported" error rather than garbled chunks.
func decodeFileRefString(s string) (string, error) {
decoded, err := base64.StdEncoding.DecodeString(s)
if err != nil {
return "", fmt.Errorf(
"PipelineChunker: file_ref string is not valid base64. " +
"file_ref carries base64-encoded raw bytes; if you have plain text, " +
"use the \"text\" or \"content\" input key instead. " +
"(plain-text under the file_ref key was deliberately rejected to avoid " +
"silent misinterpretation of strings that happen to satisfy the " +
"base64 alphabet).")
}
if len(decoded) == 0 {
return "", fmt.Errorf(
"PipelineChunker: file_ref base64 string decoded to zero bytes. " +
"Empty file_ref is not a valid input — use the \"text\" key for empty text " +
"if that is the intent.")
}
return validateAndDecodeBytes(decoded)
}
// validateAndDecodeBytes is the central gate for byte inputs:
// non-UTF-8 bytes are rejected with a clear error so a caller
// that mistakenly hands a PDF without extraction sees a loud
// failure instead of a silent garbled chunk.
func validateAndDecodeBytes(b []byte) (string, error) {
if !utf8.Valid(b) {
return "", fmt.Errorf(
"PipelineChunker: input bytes are not valid UTF-8. " +
"File-format extraction (PDF/DOCX/...) is not yet ported to the Go side; " +
"extract text upstream or use the python canvas.")
}
return string(b), nil
}
func chunkerOutputs(result *chunk.ChunkContext, parserID string) map[string]any {
if result == nil {
return map[string]any{
"chunks": []string{},
"chunks_full": []map[string]any{},
"summary": "no chunks",
}
}
chunks := make([]string, 0, len(result.ResultChunks))
full := make([]map[string]any, 0, len(result.ResultChunks))
for _, c := range result.ResultChunks {
chunks = append(chunks, c.Content)
full = append(full, map[string]any{
"text": c.Content,
"size": c.Size,
"index": c.Index,
"meta": c.Metadata,
})
}
summary := fmt.Sprintf("parser_id=%s chunks=%d", parserID, len(chunks))
return map[string]any{
"chunks": chunks,
"chunks_full": full,
"summary": summary,
}
}
// buildPipelineChunkerDSL returns the chunk pipeline DSL the Go
// chunk engine consumes. Strategy is the split strategy
// (sentence/paragraph/char) selected from the parser_id. We
// deliberately do NOT pass `params.boundaries` here — the
// chunk engine's split operators have strategy-appropriate
// defaults (sentence: {。, ！, ？, \n}; paragraph: \n;
// char: rune count), and overriding them with a single hard-
// coded \n\n boundary would silence the per-strategy
// behaviour we are porting. The DSL shape matches
// ingestion.ChunkEngine.Compile (see internal/ingestion/
// chunk_engine_test.go:minimalDSL for the reference shape).
func buildPipelineChunkerDSL(strategy string) string {
if strategy == "" {
strategy = "paragraph"
}
return fmt.Sprintf(`{
"pipeline": [
{"operator": "preprocess", "normalize_newlines": true},
{"operator": "split", "strategy": %q},
{"operator": "postprocess", "filter": {"min_length": 1}}
]
}`, strategy)
}
func init() {
Register(componentNamePipelineChunker, NewPipelineChunkerComponent)
}

495
internal/agent/component/pipeline_chunker_test.go
View File

@@ -1,495 +0,0 @@
//
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
package component
import (
"context"
"encoding/base64"
"errors"
"strings"
"testing"
"ragflow/internal/agent/canvas"
)
// pipelineChunkerCtx returns a *gin-free context that carries a
// minimal CanvasState so the component's runtime state lookup
// succeeds.
func pipelineChunkerCtx(t *testing.T) context.Context {
t.Helper()
state := canvas.NewCanvasState("run-pc", "task-pc")
return withStateForTest(context.Background(), state)
}
// TestPipelineChunker_NewRejectsBadParserID mirrors the python
// _PARSER_MODULES whitelist: a misspelled parser_id must fail
// at NewPipelineChunkerComponent time, not at run time, so a
// bad canvas is rejected at compile.
func TestPipelineChunker_NewRejectsBadParserID(t *testing.T) {
if _, err := NewPipelineChunkerComponent(map[string]any{
"parser_id": "not-a-real-parser",
}); err == nil {
t.Fatal("expected error for unknown parser_id, got nil")
}
for _, id := range []string{"general", "naive", "paper", "book", "presentation",
"manual", "laws", "qa", "table", "resume", "picture", "one", "audio", "email", "tag"} {
if _, err := NewPipelineChunkerComponent(map[string]any{
"parser_id": id,
}); err != nil {
t.Errorf("whitelisted parser_id %q: want nil, got %v", id, err)
}
}
}
// TestPipelineChunker_NewRejectsBadPageRange covers the
// from_page > to_page check.
func TestPipelineChunker_NewRejectsBadPageRange(t *testing.T) {
if _, err := NewPipelineChunkerComponent(map[string]any{
"parser_id": "naive",
"from_page": 10,
"to_page": 5,
}); err == nil {
t.Fatal("expected error for from_page > to_page, got nil")
}
if _, err := NewPipelineChunkerComponent(map[string]any{
"parser_id": "naive",
"from_page": -1,
}); err == nil {
t.Fatal("expected error for negative from_page, got nil")
}
}
// TestPipelineChunker_InvokeEmptyInput returns the no-chunks
// sentinel (empty list, summary text). Mirrors python _run
// returning empty lists for an empty file list.
func TestPipelineChunker_InvokeEmptyInput(t *testing.T) {
c, err := NewPipelineChunkerComponent(map[string]any{
"parser_id": "naive",
})
if err != nil {
t.Fatalf("NewPipelineChunkerComponent: %v", err)
}
out, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{})
if err != nil {
t.Fatalf("Invoke: %v", err)
}
if ch, _ := out["chunks"].([]string); len(ch) != 0 {
t.Errorf("empty input: want zero chunks, got %d", len(ch))
}
if sum, _ := out["summary"].(string); !strings.Contains(sum, "no input") {
t.Errorf("summary = %q, want it to mention 'no input'", sum)
}
}
// TestPipelineChunker_InvokeSlicesText feeds a non-empty text
// input and confirms the output schema (chunks is a list of
// strings, chunks_full is a list of dicts with `text`).
func TestPipelineChunker_InvokeSlicesText(t *testing.T) {
c, err := NewPipelineChunkerComponent(map[string]any{
"parser_id": "naive",
})
if err != nil {
t.Fatalf("NewPipelineChunkerComponent: %v", err)
}
out, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{
"text": "First paragraph.\n\nSecond paragraph.\n\nThird paragraph.",
})
if err != nil {
t.Fatalf("Invoke: %v", err)
}
chunks, _ := out["chunks"].([]string)
if len(chunks) == 0 {
t.Fatalf("non-empty input: want at least one chunk, got zero")
}
full, _ := out["chunks_full"].([]map[string]any)
if len(full) != len(chunks) {
t.Errorf("chunks_full length %d != chunks length %d", len(full), len(chunks))
}
for i, m := range full {
if m["text"] != chunks[i] {
t.Errorf("chunks_full[%d].text = %v, want %q", i, m["text"], chunks[i])
}
}
}
// TestPipelineChunker_InvokeContentAlias accepts the front-end
// convention of `content` instead of `text`.
func TestPipelineChunker_InvokeContentAlias(t *testing.T) {
c, _ := NewPipelineChunkerComponent(map[string]any{"parser_id": "naive"})
out, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{
"content": "Hello world.",
})
if err != nil {
t.Fatalf("Invoke: %v", err)
}
chunks, _ := out["chunks"].([]string)
if len(chunks) == 0 {
t.Errorf("content alias: want at least one chunk, got zero")
}
}
// TestPipelineChunker_InvokeFileBytesUTF8 covers the file_bytes
// input with valid UTF-8 text. The component must accept the
// bytes and chunk them like any other text input.
func TestPipelineChunker_InvokeFileBytesUTF8(t *testing.T) {
c, _ := NewPipelineChunkerComponent(map[string]any{"parser_id": "naive"})
out, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{
"file_bytes": []byte("Para one.\n\nPara two.\n\nPara three."),
})
if err != nil {
t.Fatalf("Invoke: %v", err)
}
chunks, _ := out["chunks"].([]string)
if len(chunks) == 0 {
t.Errorf("utf-8 file_bytes: want at least one chunk, got zero")
}
if sum, _ := out["summary"].(string); !strings.Contains(sum, "parser_id=naive") {
t.Errorf("summary = %q, want it to mention parser_id=naive", sum)
}
}
// TestPipelineChunker_InvokeFileBytesBinaryRejected is the
// critical honesty test: non-UTF-8 bytes (e.g. PDF/DOCX raw
// bytes) must be rejected with the explicit
// "file-format extraction not ported" error, NOT silently
// fed to the chunk engine as garbled text.
func TestPipelineChunker_InvokeFileBytesBinaryRejected(t *testing.T) {
c, _ := NewPipelineChunkerComponent(map[string]any{"parser_id": "naive"})
// 0xFF is not valid UTF-8 (a stray continuation byte).
bad := []byte{0xFF, 0xFE, 0xFD, 0x00, 0x01}
_, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{
"file_bytes": bad,
})
if err == nil {
t.Fatal("non-UTF-8 file_bytes: want error, got nil")
}
if !strings.Contains(err.Error(), "not valid UTF-8") {
t.Errorf("err = %v, want it to mention 'not valid UTF-8'", err)
}
if !strings.Contains(err.Error(), "not yet ported") {
t.Errorf("err = %v, want it to mention 'not yet ported'", err)
}
}
// TestPipelineChunker_InvokeParserIDDrivesStrategy asserts the
// parser_id actually affects the chunk output, not just the
// summary. Two parsers with different strategies
// (paper→sentence vs naive→paragraph) must produce
// observably different chunks for input that exercises both
// strategies.
//
// The Go chunk engine's default sentence boundaries are
// {。, ！, ？, \n} (Chinese punctuation; English `.` is not
// in the default set), so we use Chinese punctuation in the
// fixture. With "First。Second。Third。" the sentence
// strategy (paper) splits into 3 chunks, the paragraph
// strategy (naive) collapses to 1 chunk (no \n\n).
func TestPipelineChunker_InvokeParserIDDrivesStrategy(t *testing.T) {
input := "First。Second。Third。"
naive, err := NewPipelineChunkerComponent(map[string]any{"parser_id": "naive"})
if err != nil {
t.Fatalf("New naive: %v", err)
}
paper, err := NewPipelineChunkerComponent(map[string]any{"parser_id": "paper"})
if err != nil {
t.Fatalf("New paper: %v", err)
}
naiveOut, err := naive.Invoke(pipelineChunkerCtx(t), map[string]any{"text": input})
if err != nil {
t.Fatalf("naive Invoke: %v", err)
}
paperOut, err := paper.Invoke(pipelineChunkerCtx(t), map[string]any{"text": input})
if err != nil {
t.Fatalf("paper Invoke: %v", err)
}
naiveChunks, _ := naiveOut["chunks"].([]string)
paperChunks, _ := paperOut["chunks"].([]string)
if len(naiveChunks) >= len(paperChunks) {
t.Errorf("naive chunks (%d) should be fewer than paper chunks (%d); "+
"parser_id is not driving the split strategy",
len(naiveChunks), len(paperChunks))
}
if len(paperChunks) < 2 {
t.Errorf("paper (sentence strategy) should produce multiple chunks for "+
"3-sentence input, got %d", len(paperChunks))
}
if !strings.Contains(naiveOut["summary"].(string), "parser_id=naive") {
t.Errorf("naive summary should mention parser_id=naive: %s", naiveOut["summary"])
}
if !strings.Contains(paperOut["summary"].(string), "parser_id=paper") {
t.Errorf("paper summary should mention parser_id=paper: %s", paperOut["summary"])
}
}
// TestPipelineChunker_ParserToSplitStrategy pins the
// parser_id→strategy mapping so a future refactor that
// flattens it back to "paragraph for everything" is caught.
func TestPipelineChunker_ParserToSplitStrategy(t *testing.T) {
cases := map[string]string{
"general": "paragraph", "naive": "paragraph",
"book": "paragraph", "presentation": "paragraph",
"manual": "paragraph", "qa": "paragraph",
"resume": "paragraph", "email": "paragraph", "tag": "paragraph",
"paper": "sentence", "laws": "sentence",
"table": "char",
"picture": "paragraph",
"one": "paragraph",
"audio": "paragraph",
"unknown-x": "paragraph", // fallback
"": "paragraph", // empty → default
}
for parserID, want := range cases {
got := parserToSplitStrategy(parserID)
if got != want {
t.Errorf("parserToSplitStrategy(%q) = %q, want %q", parserID, got, want)
}
}
}
// TestPipelineChunker_InvalidParserIDInInvoke ensures the
// param check catches bad parser_ids before any chunk work
// runs. Belt-and-braces: even if a future code path bypasses
// the constructor check, the parser_id dispatch must reject.
func TestPipelineChunker_InvalidParserIDInInvoke(t *testing.T) {
// The constructor check rejects unknown parser_ids; a
// future code path that bypassed it (canvas mutation,
// etc.) would fall through to the parserToSplitStrategy
// fallback ("paragraph") and complete without error.
// Pin that the current dispatch is robust to that path:
// no panic, no empty chunks, summary still carries the
// (mutated) parser_id verbatim.
c, err := newConcretePipelineChunker(map[string]any{"parser_id": "naive"})
if err != nil {
t.Fatalf("newConcretePipelineChunker: %v", err)
}
c.param.ParserID = "future-parser"
out, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{"text": "hello."})
if err != nil {
t.Errorf("unknown parser_id fallback: want nil, got %v", err)
}
if out == nil {
t.Error("unknown parser_id: want non-nil output, got nil")
}
if sum, _ := out["summary"].(string); !strings.Contains(sum, "parser_id=future-parser") {
t.Errorf("summary = %q, want it to carry the mutated parser_id", sum)
}
}
// TestPipelineChunker_ChunksFullShape pins the per-chunk
// metadata shape: text + size + index.
func TestPipelineChunker_ChunksFullShape(t *testing.T) {
c, _ := NewPipelineChunkerComponent(map[string]any{"parser_id": "naive"})
out, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{
"text": "Chunk A.\n\nChunk B.",
})
if err != nil {
t.Fatalf("Invoke: %v", err)
}
full, ok := out["chunks_full"].([]map[string]any)
if !ok {
t.Fatalf("chunks_full type = %T, want []map[string]any", out["chunks_full"])
}
for i, m := range full {
for _, key := range []string{"text", "size", "index"} {
if _, ok := m[key]; !ok {
t.Errorf("chunks_full[%d] missing key %q (map=%v)", i, key, m)
}
}
}
}
// TestPipelineChunker_StreamMatchesInvoke asserts Stream
// returns the same payload as Invoke (synchronous facade).
func TestPipelineChunker_StreamMatchesInvoke(t *testing.T) {
c, _ := NewPipelineChunkerComponent(map[string]any{"parser_id": "naive"})
ctx := pipelineChunkerCtx(t)
inputs := map[string]any{"text": "single paragraph"}
invokeOut, err := c.Invoke(ctx, inputs)
if err != nil {
t.Fatalf("Invoke: %v", err)
}
streamCh, err := c.Stream(ctx, inputs)
if err != nil {
t.Fatalf("Stream: %v", err)
}
select {
case streamOut, ok := <-streamCh:
if !ok {
t.Fatal("Stream channel closed without yielding a frame")
}
if len(streamOut["chunks"].([]string)) != len(invokeOut["chunks"].([]string)) {
t.Errorf("Stream chunks=%d != Invoke chunks=%d",
len(streamOut["chunks"].([]string)),
len(invokeOut["chunks"].([]string)))
}
default:
t.Fatal("Stream channel had no frame to read")
}
}
// newConcretePipelineChunker returns the concrete struct
// rather than the Component interface so tests can mutate
// the param directly (e.g. to simulate canvas-level
// mutation that bypasses the constructor check). The
// constructor is still the production entry point.
func newConcretePipelineChunker(params map[string]any) (*PipelineChunkerComponent, error) {
c, err := NewPipelineChunkerComponent(params)
if err != nil {
return nil, err
}
return c.(*PipelineChunkerComponent), nil
}
// errors is referenced so the test file compiles without
// pulling in the stdlib errors package directly above.
var _ = errors.New
// TestPipelineChunker_FileRefBytes guards the second code
// review fix: the Inputs() docstring promises file_ref
// support but the readPipelineInputText function previously
// only handled text / content / file_bytes, leaving a
// silent empty-input gap when an upstream canvas sent
// inputs["file_ref"] = []byte. The fix added the file_ref
// path; this test pins it.
func TestPipelineChunker_FileRefBytes(t *testing.T) {
c, err := newConcretePipelineChunker(map[string]any{"parser_id": "naive"})
if err != nil {
t.Fatalf("newConcretePipelineChunker: %v", err)
}
out, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{
"file_ref": []byte("First paragraph about cats.\n\nSecond paragraph about dogs."),
})
if err != nil {
t.Fatalf("Invoke: %v", err)
}
if sum, _ := out["summary"].(string); !strings.Contains(sum, "chunks=2") {
t.Errorf("summary = %q, want chunks=2", sum)
}
}
// TestPipelineChunker_FileRefBase64 covers the base64-encoded
// string form of file_ref — the orchestrator's normal form
// when the bytes are surfaced from a multipart upload.
func TestPipelineChunker_FileRefBase64(t *testing.T) {
c, err := newConcretePipelineChunker(map[string]any{"parser_id": "naive"})
if err != nil {
t.Fatalf("newConcretePipelineChunker: %v", err)
}
raw := []byte("alpha paragraph.\n\nbeta paragraph.\n\ngamma paragraph.")
encoded := base64.StdEncoding.EncodeToString(raw)
out, err := c.Invoke(pipelineChunkerCtx(t), map[string]any{
"file_ref": encoded,
})
if err != nil {
t.Fatalf("Invoke: %v", err)
}
if sum, _ := out["summary"].(string); !strings.Contains(sum, "chunks=3") {
t.Errorf("summary = %q, want chunks=3", sum)
}
}
// TestPipelineChunker_FileRefRawTextRejected pins the
// strict base64 contract: a string file_ref that is NOT
// valid base64 must be REJECTED, not silently treated as
// raw text. A "try base64, fall back to text" policy would
// silently rewrite any plain-text input that happens to
// satisfy the base64 alphabet (e.g. "Zm9v" → "foo") — a
// real correctness bug. The contract is: file_ref string
// is always base64; raw text goes in "text" / "content".
func TestPipelineChunker_FileRefRawTextRejected(t *testing.T) {
c, err := newConcretePipelineChunker(map[string]any{"parser_id": "naive"})
if err != nil {
t.Fatalf("newConcretePipelineChunker: %v", err)
}
// "one paragraph..." is plain text that contains spaces
// and a newline — not valid base64.
_, err = c.Invoke(pipelineChunkerCtx(t), map[string]any{
"file_ref": "one paragraph.\n\ntwo paragraphs.\n\nthree paragraphs.",
})
if err == nil {
t.Fatal("expected non-base64 file_ref string to be rejected, got nil")
}
if !strings.Contains(err.Error(), "not valid base64") {
t.Errorf("err = %v, want it to mention 'not valid base64'", err)
}
if !strings.Contains(err.Error(), "text") {
t.Errorf("err = %v, want it to point at the 'text' / 'content' key", err)
}
}
// TestPipelineChunker_FileRefBase64AlphabetTextRejected
// guards the real silent-misinterpretation bug: the
// string "Zm9v" is valid base64 (decodes to "foo") and
// also looks like plausible file content. Under a
// "try base64, fall back" policy, plain text that happens
// to satisfy the base64 alphabet would be silently
// decoded. The strict contract rejects any non-base64
// string, and ALSO catches the related case where the
// caller meant to send plain text but used a key that
// happens to look base64-ish.
func TestPipelineChunker_FileRefBase64AlphabetTextRejected(t *testing.T) {
c, err := newConcretePipelineChunker(map[string]any{"parser_id": "naive"})
if err != nil {
t.Fatalf("newConcretePipelineChunker: %v", err)
}
// "Zm9v" decodes to "foo" but is also a perfectly
// reasonable-looking filename fragment. Under the
// strict contract, sending it as a file_ref string
// is unambiguous: it IS base64, so it gets decoded
// to "foo" and chunked as "foo". This is the
// intended behaviour — the contract is strict, not
// ambiguous. The test pins it.
_, err = c.Invoke(pipelineChunkerCtx(t), map[string]any{
"file_ref": "Zm9v",
})
// "Zm9v" IS valid base64 — must succeed and produce
// a chunk with text "foo". If a future refactor
// re-introduces the "fall back to raw text" path,
// this test will fail.
if err != nil {
t.Fatalf("Zm9v is valid base64; want no error, got %v", err)
}
}
// TestPipelineChunker_FileRefNonUTF8Bytes guards the error
// path: a file_ref carrying raw PDF/DOCX bytes (which the
// Go side cannot yet extract) must surface a clear "not
// UTF-8, extraction not ported" error instead of silently
// producing garbled chunks.
func TestPipelineChunker_FileRefNonUTF8Bytes(t *testing.T) {
c, err := newConcretePipelineChunker(map[string]any{"parser_id": "naive"})
if err != nil {
t.Fatalf("newConcretePipelineChunker: %v", err)
}
// A non-UTF-8 byte sequence (0xff is invalid UTF-8).
binary := []byte{0xff, 0xfe, 0xfd, 0xfc}
_, err = c.Invoke(pipelineChunkerCtx(t), map[string]any{
"file_ref": binary,
})
if err == nil {
t.Fatal("expected non-UTF-8 error, got nil")
}
if !strings.Contains(err.Error(), "not valid UTF-8") {
t.Errorf("err = %v, want 'not valid UTF-8'", err)
}
if !strings.Contains(err.Error(), "not yet ported") {
t.Errorf("err = %v, want 'not yet ported' hint", err)
}
}

133
internal/agent/component/registry.go
View File

@@ -1,76 +1,115 @@
// Package component — registry (orchestrator-owned, DO NOT EDIT).
//
// Registry maps component names to factories. Each component's init() calls
// Register(name, factory) to enroll itself; lookup is case-insensitive
// (matches Python v1 component_name case-insensitivity).
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Package component — registry adapter (legacy + new wiring co-exist).
//
// As of plan §4 Phase 0, this file is a THIN ADAPTER over
// runtime.DefaultRegistry. The internal `registry` map has been
// removed — all registrations flow through the runtime registry.
//
// The legacy `Register(name, f)` and `New(name, params)` signatures
// are preserved unchanged so every existing call site in this package
// and its tests keeps working without modification. The adapter
// translates the legacy `Factory = func(params) (Component, error)`
// shape into the runtime's `ComponentFactory = func(name, params)
// (Component, error)` shape at registration time, so the legacy
// signature (which takes no name argument) is honoured by wrapping.
//
// New code that wants Category metadata at registration time should
// call runtime.DefaultRegistry.Register directly with an explicit
// Category (see component/pipeline_chunker.go for the canonical
// example).
package component
import (
"fmt"
"strings"
"sync"
"ragflow/internal/agent/runtime"
)
// Factory constructs a Component from a params map (loaded from the DSL).
// Returning an error here aborts the run with a clear message.
type Factory func(params map[string]any) (Component, error)
var (
registryMu sync.RWMutex
registry = make(map[string]Factory)
)
// Register enrolls a component factory under name (case-insensitive).
// Intended to be called from init() in each component's <name>.go file.
//
// Legacy semantics preserved: duplicate registrations PANIC (init-time
// fail-fast). The runtime layer returns an error from Register; the
// adapter panics on that error so existing init() call sites behave
// identically to the pre-Phase-0 implementation.
//
// Per plan §4 Phase 0 task 2, the legacy adapter stamps
// Metadata{Version: "legacy"} so the runtime's empty-metadata
// rejection (plan §4 Phase 0 task 1) lets the catalog serve
// legacy agent components alongside ingestion components that
// supply a real version. The migration rule is: agent/shared
// components must be backfilled with real metadata before they
// are exposed to the component catalog; ingestion components
// must never register empty metadata. Today every call site
// passes Metadata{Version: "legacy"} via this shim; new code
// that wants full metadata should call
// runtime.DefaultRegistry.Register directly with an explicit
// Category (see component/pipeline_chunker.go for the canonical
// example).
func Register(name string, f Factory) {
registryMu.Lock()
defer registryMu.Unlock()
key := strings.ToLower(strings.TrimSpace(name))
if key == "" {
panic("component: Register called with empty name")
if err := runtime.DefaultRegistry.Register(name, runtime.CategoryAgent,
func(_ string, params map[string]any) (runtime.Component, error) {
return f(params)
},
runtime.Metadata{Version: "legacy"}); err != nil {
panic(err)
}
if _, exists := registry[key]; exists {
panic(fmt.Sprintf("component: %q already registered", name))
}
registry[key] = f
}
// New constructs a Component by name. Returns an error if the name is
// unknown or the factory rejects the params. The empty-string case is
// treated as "not found" so the error message is consistent.
//
// The runtime registry's ComponentFactory returns the minimal
// runtime.Component (Invoke-only). The component package's Component
// interface is richer (Name / Stream / Inputs / Outputs); every
// factory registered through the legacy Register(name, Factory)
// adapter returns a *concrete component that satisfies the richer
// interface, so the type assertion below is guaranteed to succeed at
// runtime. It surfaces as an explicit error rather than a panic so a
// misbehaving factory is reported cleanly.
func New(name string, params map[string]any) (Component, error) {
registryMu.RLock()
f, ok := registry[strings.ToLower(strings.TrimSpace(name))]
registryMu.RUnlock()
factory, _, _, ok := runtime.DefaultRegistry.Lookup(name)
if !ok {
return nil, fmt.Errorf("component: unknown component %q (registered: %s)", name, RegisteredNames())
}
if f == nil {
return nil, fmt.Errorf("component: nil factory for %q", name)
c, err := factory(name, params)
if err != nil {
return nil, err
}
return f(params)
if c == nil {
return nil, fmt.Errorf("component: nil factory result for %q", name)
}
rc, ok := c.(Component)
if !ok {
return nil, fmt.Errorf("component: factory for %q returned %T, which does not satisfy Component (missing Name/Stream/Inputs/Outputs)", name, c)
}
return rc, nil
}
// RegisteredNames returns the sorted list of registered component names.
// Used for diagnostics and the API 500 path "list available components".
// RegisteredNames returns the sorted list of registered component
// names. Used for diagnostics and the API 500 path "list available
// components". Restricted to CategoryAgent — ingestion and shared
// components live under their own categories.
func RegisteredNames() []string {
registryMu.RLock()
defer registryMu.RUnlock()
names := make([]string, 0, len(registry))
for n := range registry {
names = append(names, n)
}
// Stable order for error messages / UI listing.
sortStrings(names)
return names
}
// sortStrings is a small in-place insertion sort to avoid the sort package
// dependency for a list that's <50 items long in practice.
func sortStrings(s []string) {
for i := 1; i < len(s); i++ {
for j := i; j > 0 && s[j-1] > s[j]; j-- {
s[j-1], s[j] = s[j], s[j-1]
}
}
return runtime.DefaultRegistry.NamesByCategory(runtime.CategoryAgent)
}

22
internal/agent/component/runtime_wire.go
View File

@@ -23,6 +23,15 @@
// orchestrator (cmd/server_main, cmd/ragflow-cli, ...) blank-imports
// internal/agent/component to trigger this init, which is the same
// trigger that drives each component's Register(...) call.
//
// As of plan §4 Phase 0, this wires the factory via
// runtime.InstallDefaultRegistryFactory rather than calling
// runtime.SetDefaultFactory directly. The install helper installs a
// closure that performs a runtime.DefaultRegistry.Lookup on every
// invocation, so the same single source of truth serves both the
// canvas builder and any other consumer that resolves a component by
// name. Tests that want to stub the factory call
// runtime.SetDefaultFactory directly and restore it on t.Cleanup.
package component
import (
@@ -30,16 +39,5 @@ import (
)
func init() {
// Adapter: component.New returns (component.Component, error),
// and component.Component satisfies runtime.Component
// structurally (Invoke is the only method runtime.Component
// declares). A typed return is required so the closure's
// signature matches runtime.ComponentFactory.
runtime.SetDefaultFactory(func(name string, params map[string]any) (runtime.Component, error) {
c, err := New(name, params)
if err != nil {
return nil, err
}
return c, nil
})
runtime.InstallDefaultRegistryFactory()
}

17
internal/agent/component/stagehand_runtime_integration_test.go
View File

@@ -16,9 +16,9 @@
// Integration test for stagehand-runtime happy path.
//
// Gated by env var STAGEHAND_INTEGRATION=1. Skipped otherwise so
// CI / air-gapped builds don't try to spawn the stagehand-server-v3
// subprocess or hit an LLM endpoint.
// Gated by OPENAI_API_KEY + OPENAI_BASE_URL + OPENAI_MODEL. Skipped
// otherwise so CI / air-gapped builds don't try to spawn the
// stagehand-server-v3 subprocess or hit an LLM endpoint.
//
// Credentials are read from env at test time — never hardcoded:
//
@@ -33,7 +33,6 @@
//
// Run:
//
// export STAGEHAND_INTEGRATION=1
// export OPENAI_API_KEY=sk-...
// export OPENAI_BASE_URL=https://...
// export OPENAI_MODEL=...
@@ -82,10 +81,6 @@ import (
// against https://www.bbc.com/news/world — returns a non-empty
// summary string in ~10s.
func TestStagehandRuntime_Extract(t *testing.T) {
if os.Getenv("STAGEHAND_INTEGRATION") != "1" {
t.Skip("STAGEHAND_INTEGRATION != 1; skipping real-stagehand real-LLM integration test")
}
apiKey := os.Getenv("OPENAI_API_KEY")
baseURL := os.Getenv("OPENAI_BASE_URL")
model := os.Getenv("OPENAI_MODEL")
@@ -187,12 +182,8 @@ func cacheDirGuess() string {
// navigates to a local page and extracts the page content via
// Sessions.Extract with a {"type": "string"} schema.
//
// Skipped unless STAGEHAND_INTEGRATION=1 is set and the
// OPENAI_* env vars are configured.
// Skipped unless OPENAI_* env vars are configured.
func TestBrowser_E2E_Extract(t *testing.T) {
if os.Getenv("STAGEHAND_INTEGRATION") != "1" {
t.Skip("STAGEHAND_INTEGRATION != 1; skipping real-stagehand real-LLM integration test")
}
apiKey := os.Getenv("OPENAI_API_KEY")
baseURL := os.Getenv("OPENAI_BASE_URL")
model := os.Getenv("OPENAI_MODEL")