fix(ingestion): document and validate extractor timeout env (#21119)

This commit is contained in:
jay77721 authored and GitHub committed 2026-10-19 13:07:23 +08:00
1 parent 73c2b24079
commit 1d2d049702
4 files changed
+27 -13

No files matched your search

+5
View File
@@ -367,6 +367,11 @@ DOC_BULK_SIZE=${DOC_BULK_SIZE:-4}
# Defaults to 16 if EMBEDDING_BATCH_SIZE is not set in the environment.
EMBEDDING_BATCH_SIZE=${EMBEDDING_BATCH_SIZE:-16}
# Extractor LLM call timeout in seconds, including retries and backoff.
# Applies to keyword, question, summary, and metadata extraction.
# Default: 180. Invalid or non-positive values use the default.
# EXTRACTOR_CHAT_TIMEOUT=180
# Connector ingest: documents (e.g. one MySQL row = one file) written per batch.
# Defaults to 2. Raise this for large RDBMS syncs so fewer write round-trips hit MySQL.
# INDEX_BATCH_SIZE=32
+4
View File
@@ -255,6 +255,10 @@ const (
EnvGoogleDriveWebOAuthRedirectURI = "GOOGLE_DRIVE_WEB_OAUTH_REDIRECT_URI"
EnvSpacyModelDir = "SPACY_MODEL_DIR"
// EnvExtractorChatTimeout sets the Extractor LLM call timeout in seconds,
// including retries and backoff. It does not apply to tag extraction.
EnvExtractorChatTimeout = "EXTRACTOR_CHAT_TIMEOUT"
// EnvDeepDocModelDir points the in-process (Go) DeepDoc backend at the
// model snapshot (see common.DeepDocModelFiles); mirrors
// the RAGFlow default model dir (internal/rag/res/deepdoc).
+5 -8
View File
@@ -39,6 +39,7 @@ import (
"encoding/json"
"errors"
"fmt"
"math"
"os"
"regexp"
"sort"
@@ -66,20 +67,16 @@ import (
const componentNameExtractor = "Extractor"
// defaultExtractorChatTimeout bounds one LLM chat call. It is a hung-call
// ceiling, not a pacing target: a fast model answers in seconds and never
// approaches it, so per-model speed variance needs no tuning — only a hung
// provider burns the full budget. Larger than the tag path's 60s because a
// thinking model legitimately needs longer on open-ended prompts, and the
// driver-level 20m stays as the absolute backstop.
// defaultExtractorChatTimeout bounds one Extractor LLM call, including
// retries and backoff. Tag extraction has its own timeout.
const defaultExtractorChatTimeout = 180 * time.Second
// extractorChatTimeout resolves the per-call LLM budget from
// EXTRACTOR_CHAT_TIMEOUT (seconds), falling back to
// defaultExtractorChatTimeout on unset/invalid values.
func extractorChatTimeout() time.Duration {
if v := os.Getenv("EXTRACTOR_CHAT_TIMEOUT"); v != "" {
if n, err := strconv.Atoi(strings.TrimSpace(v)); err == nil && n > 0 {
if v := os.Getenv(common.EnvExtractorChatTimeout); v != "" {
if n, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64); err == nil && n > 0 && n <= math.MaxInt64/int64(time.Second) {
return time.Duration(n) * time.Second
}
}
@@ -24,6 +24,7 @@ import (
"testing"
"time"
"ragflow/internal/common"
"ragflow/internal/ingestion/component/schema"
)
@@ -71,7 +72,7 @@ func keywordsOnlyExtractor() *ExtractorComponent {
// TestExtractorChatTimeoutSkipsSlowChunk: one hung chunk must not fail the
// document. The healthy chunk keeps its extraction, the hung one is skipped.
func TestExtractorChatTimeoutSkipsSlowChunk(t *testing.T) {
t.Setenv("EXTRACTOR_CHAT_TIMEOUT", "1")
t.Setenv(common.EnvExtractorChatTimeout, "1")
stub := withRouteStubChatInvoker(t)
c := keywordsOnlyExtractor()
@@ -102,7 +103,7 @@ func TestExtractorChatTimeoutSkipsSlowChunk(t *testing.T) {
// TestExtractorChatTimeoutAllHangFails: every chunk hung means the model is
// down — the run must fail loudly, not silently produce empty extractions.
func TestExtractorChatTimeoutAllHangFails(t *testing.T) {
t.Setenv("EXTRACTOR_CHAT_TIMEOUT", "1")
t.Setenv(common.EnvExtractorChatTimeout, "1")
withRouteStubChatInvoker(t)
c := keywordsOnlyExtractor()
@@ -123,7 +124,7 @@ func TestExtractorChatTimeoutAllHangFails(t *testing.T) {
// TestExtractorParentCancelPropagates: task stop must still fail fast, never
// degrade into per-chunk skips.
func TestExtractorParentCancelPropagates(t *testing.T) {
t.Setenv("EXTRACTOR_CHAT_TIMEOUT", "1")
t.Setenv(common.EnvExtractorChatTimeout, "1")
withRouteStubChatInvoker(t)
ctx, cancel := context.WithCancel(t.Context())
@@ -150,12 +151,19 @@ func TestExtractorChatTimeoutEnv(t *testing.T) {
}{
{name: "unset falls back to default", value: "", want: defaultExtractorChatTimeout},
{name: "override wins", value: "1", want: time.Second},
{name: "whitespace falls back", value: " \t", want: defaultExtractorChatTimeout},
{name: "trimmed override wins", value: " 240 ", want: 240 * time.Second},
{name: "duration suffix falls back", value: "180s", want: defaultExtractorChatTimeout},
{name: "fractional seconds fall back", value: "1.5", want: defaultExtractorChatTimeout},
{name: "maximum duration seconds", value: "9223372036", want: 9223372036 * time.Second},
{name: "duration overflow falls back", value: "9223372037", want: defaultExtractorChatTimeout},
{name: "integer overflow falls back", value: "9223372036854775808", want: defaultExtractorChatTimeout},
{name: "zero falls back", value: "0", want: defaultExtractorChatTimeout},
{name: "negative falls back", value: "-5", want: defaultExtractorChatTimeout},
{name: "non-numeric falls back", value: "abc", want: defaultExtractorChatTimeout},
} {
t.Run(tt.name, func(t *testing.T) {
t.Setenv("EXTRACTOR_CHAT_TIMEOUT", tt.value)
t.Setenv(common.EnvExtractorChatTimeout, tt.value)
if got := extractorChatTimeout(); got != tt.want {
t.Errorf("extractorChatTimeout() = %v, want %v", got, tt.want)
}
@@ -197,7 +205,7 @@ func withSummaryHangStub(t *testing.T) {
// summaries hang for every chunk. The phase must succeed and keep the
// questions instead of failing as all-timed-out.
func TestExtractorPartialTimeoutKeepsCompleted(t *testing.T) {
t.Setenv("EXTRACTOR_CHAT_TIMEOUT", "1")
t.Setenv(common.EnvExtractorChatTimeout, "1")
withSummaryHangStub(t)
c := &ExtractorComponent{Param: schema.ExtractorParam{