From 7f08d9295816e1a06d76e50902de3569b7d7e862 Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 11 May 2026 11:11:21 +0800 Subject: [PATCH] feat(browser-fetch): add lightpanda-backed fetch service and knowledge URL fallback Adds a new browser-fetch microservice that wraps Lightpanda for rendering JS-heavy pages, and wires it into the knowledge URL parser as a fallback when the upstream link reader returns 403/429/5xx for an allow-listed domain (e.g. zhuanlan.zhihu.com). Includes config/env plumbing, hot-reload diff support, and full deploy assets (Dockerfile target, docker-compose, k3s manifests, image-build/package scripts). --- deploy/.env.example | 1 + deploy/config.yaml | 20 +- deploy/docker-compose.offline.yaml | 5 + deploy/docker-compose.yaml | 28 + deploy/k3s/README.md | 4 +- deploy/k3s/apps.yaml | 94 ++ deploy/k3s/config/config.yaml | 23 +- deploy/k3s/kustomization.yaml | 3 + deploy/k3s/secrets.yaml | 4 + deploy/scripts/gitea-registry-images.sh | 11 +- deploy/scripts/package.sh | 4 +- server/Dockerfile | 16 + server/cmd/browser-fetch/main.go | 877 ++++++++++++++++++ server/configs/config.local.yaml.example | 20 + server/configs/config.yaml | 20 + server/internal/shared/config/config.go | 110 +++ server/internal/shared/config/config_test.go | 87 ++ server/internal/shared/config/reload.go | 6 + .../tenant/app/browser_fetch_queue.go | 236 +++++ .../internal/tenant/app/knowledge_service.go | 4 +- .../tenant/app/knowledge_url_parser.go | 193 +++- server/internal/tenant/app/runtime_config.go | 12 + 22 files changed, 1770 insertions(+), 8 deletions(-) create mode 100644 server/cmd/browser-fetch/main.go create mode 100644 server/internal/tenant/app/browser_fetch_queue.go diff --git a/deploy/.env.example b/deploy/.env.example index d203a0a..0346b57 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -14,6 +14,7 @@ JWT_SECRET=geo-web-Aa123456-change-me-in-production OPS_JWT_SECRET=geo-ops-Aa123456-change-me-in-production OPS_DEFAULT_ADMIN_PASSWORD=Aa123456 SCHEDULER_INTERNAL_METRICS_TOKEN=geo-scheduler-Aa123456-change-me-in-production +BROWSER_FETCH_TOKEN=geo-browser-fetch-Aa123456-change-me-in-production POSTGRES_PASSWORD=geo_dev # ─── Optional: MinIO credentials ───────────────────────────────────────────── diff --git a/deploy/config.yaml b/deploy/config.yaml index 32cb9f1..d32b81d 100644 --- a/deploy/config.yaml +++ b/deploy/config.yaml @@ -215,4 +215,22 @@ generation: task_recovery_batch_size: 100 task_queued_stale_after: 2m task_max_attempts: 3 - + +browser_fetch: + enabled: true + provider: lightpanda + service_url: http://browser-fetch:8082 + queue_size: 64 + worker_concurrency: 2 + request_timeout: 45s + cache_ttl: 6h + allowed_domains: + - zhuanlan.zhihu.com + - zhihu.com + trigger_status: + - 403 + - 429 + - 500 + - 502 + - 503 + - 504 diff --git a/deploy/docker-compose.offline.yaml b/deploy/docker-compose.offline.yaml index c9448da..2aa895b 100644 --- a/deploy/docker-compose.offline.yaml +++ b/deploy/docker-compose.offline.yaml @@ -18,6 +18,11 @@ services: pull_policy: never build: !reset null + browser-fetch: + image: geo-rankly/browser-fetch:${IMAGE_TAG:-latest} + pull_policy: never + build: !reset null + worker-generate: image: geo-rankly/worker-generate:${IMAGE_TAG:-latest} pull_policy: never diff --git a/deploy/docker-compose.yaml b/deploy/docker-compose.yaml index 66f4eff..eecd935 100644 --- a/deploy/docker-compose.yaml +++ b/deploy/docker-compose.yaml @@ -16,6 +16,7 @@ x-app-env: &app-env JWT_ACCESS_TTL: ${JWT_ACCESS_TTL:-15m} JWT_REFRESH_TTL: ${JWT_REFRESH_TTL:-720h} SCHEDULER_INTERNAL_METRICS_TOKEN: ${SCHEDULER_INTERNAL_METRICS_TOKEN:-} + BROWSER_FETCH_TOKEN: ${BROWSER_FETCH_TOKEN:-} OBJECT_STORAGE_ACCESS_KEY: ${MINIO_ROOT_USER:-minioadmin} OBJECT_STORAGE_SECRET_KEY: ${MINIO_ROOT_PASSWORD:-minioadmin} @@ -261,6 +262,33 @@ services: retries: 5 start_period: 10s + browser-fetch: + image: ${IMAGE_REGISTRY:-ghcr.io/your-org/geo-rankly}/browser-fetch:${IMAGE_TAG:-latest} + container_name: geo-browser-fetch + build: + context: ../server + dockerfile: Dockerfile + target: browser-fetch + args: + SERVICE: browser-fetch + restart: unless-stopped + environment: + BROWSER_FETCH_TOKEN: ${BROWSER_FETCH_TOKEN:-geo-browser-fetch-change-me} + BROWSER_FETCH_ALLOWED_DOMAINS: zhuanlan.zhihu.com,zhihu.com + BROWSER_FETCH_QUEUE_SIZE: 128 + BROWSER_FETCH_WORKER_CONCURRENCY: 2 + BROWSER_FETCH_PER_DOMAIN_CONCURRENCY: 1 + BROWSER_FETCH_REQUEST_TIMEOUT: 45s + BROWSER_FETCH_CACHE_TTL: 6h + LIGHTPANDA_DRIVER: fetch + LIGHTPANDA_FETCH_ARGS: --dump markdown + healthcheck: + test: ["/app/service", "-healthcheck"] + interval: 10s + timeout: 5s + retries: 5 + start_period: 10s + worker-generate: image: ${IMAGE_REGISTRY:-ghcr.io/your-org/geo-rankly}/worker-generate:${IMAGE_TAG:-latest} container_name: geo-worker-generate diff --git a/deploy/k3s/README.md b/deploy/k3s/README.md index 884f62f..cdf7b2a 100644 --- a/deploy/k3s/README.md +++ b/deploy/k3s/README.md @@ -4,7 +4,7 @@ - infra: `postgres`, `monitoring-postgres`, `pgbouncer`, `monitoring-pgbouncer`, `rabbitmq`, `redis`, `qdrant`, `minio` - jobs: `migrate`, `minio-init` -- apps: `tenant-api`, `worker-generate`, `kol-assist-worker`, `scheduler`, `ops-api`, `frontend`, `ops-web` +- apps: `tenant-api`, `browser-fetch`, `worker-generate`, `kol-assist-worker`, `scheduler`, `ops-api`, `frontend`, `ops-web` - ingress: `geo-rankly.local` -> `frontend`, `ops.geo-rankly.local` -> `ops-web` ## 1. 准备镜像 @@ -34,6 +34,7 @@ sudo k3s ctr images import images.tar - `OPS_JWT_SECRET` - `OPS_DEFAULT_ADMIN_PASSWORD` - `SCHEDULER_INTERNAL_METRICS_TOKEN` +- `BROWSER_FETCH_TOKEN` - `POSTGRES_PASSWORD` / `RABBITMQ_DEFAULT_PASS` / `MINIO_ROOT_*` - `config/config.yaml` 里的 LLM / SiliconFlow API key - `config.local.yaml` 里对应的数据库、RabbitMQ、MinIO、JWT 配置 @@ -131,6 +132,7 @@ kubectl -n geo-rankly port-forward svc/minio 9001:9001 kubectl -n geo-rankly get all kubectl -n geo-rankly describe pod kubectl -n geo-rankly logs deploy/tenant-api -f +kubectl -n geo-rankly logs deploy/browser-fetch -f kubectl -n geo-rankly logs deploy/scheduler -f kubectl -n geo-rankly logs deploy/worker-generate -f kubectl -n geo-rankly logs deploy/ops-api -f diff --git a/deploy/k3s/apps.yaml b/deploy/k3s/apps.yaml index 0a280e1..b82eca0 100644 --- a/deploy/k3s/apps.yaml +++ b/deploy/k3s/apps.yaml @@ -1,5 +1,89 @@ apiVersion: v1 kind: Service +metadata: + name: browser-fetch + labels: + app.kubernetes.io/name: browser-fetch +spec: + selector: + app.kubernetes.io/name: browser-fetch + ports: + - name: http + port: 8082 + targetPort: http +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: browser-fetch + labels: + app.kubernetes.io/name: browser-fetch +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: browser-fetch + template: + metadata: + labels: + app.kubernetes.io/name: browser-fetch + spec: + containers: + - name: browser-fetch + image: geo-rankly/browser-fetch:latest + imagePullPolicy: IfNotPresent + ports: + - name: http + containerPort: 8082 + env: + - name: TZ + value: Asia/Shanghai + - name: BROWSER_FETCH_TOKEN + valueFrom: + secretKeyRef: + name: geo-rankly-app-secret + key: BROWSER_FETCH_TOKEN + - name: BROWSER_FETCH_ALLOWED_DOMAINS + value: zhuanlan.zhihu.com,zhihu.com + - name: BROWSER_FETCH_QUEUE_SIZE + value: "128" + - name: BROWSER_FETCH_WORKER_CONCURRENCY + value: "2" + - name: BROWSER_FETCH_PER_DOMAIN_CONCURRENCY + value: "1" + - name: BROWSER_FETCH_REQUEST_TIMEOUT + value: 45s + - name: BROWSER_FETCH_CACHE_TTL + value: 6h + - name: LIGHTPANDA_DRIVER + value: fetch + - name: LIGHTPANDA_FETCH_ARGS + value: --dump markdown + readinessProbe: + exec: + command: + - /app/service + - -healthcheck + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 3 + livenessProbe: + httpGet: + path: /health/live + port: http + initialDelaySeconds: 15 + periodSeconds: 20 + timeoutSeconds: 3 + resources: + requests: + cpu: 500m + memory: 512Mi + limits: + cpu: "2" + memory: 2Gi +--- +apiVersion: v1 +kind: Service metadata: name: tenant-api labels: @@ -60,6 +144,11 @@ spec: secretKeyRef: name: geo-rankly-app-secret key: MINIO_ROOT_PASSWORD + - name: BROWSER_FETCH_TOKEN + valueFrom: + secretKeyRef: + name: geo-rankly-app-secret + key: BROWSER_FETCH_TOKEN - name: SCHEDULER_INTERNAL_METRICS_TOKEN valueFrom: secretKeyRef: @@ -152,6 +241,11 @@ spec: secretKeyRef: name: geo-rankly-app-secret key: MINIO_ROOT_PASSWORD + - name: BROWSER_FETCH_TOKEN + valueFrom: + secretKeyRef: + name: geo-rankly-app-secret + key: BROWSER_FETCH_TOKEN resources: requests: cpu: 200m diff --git a/deploy/k3s/config/config.yaml b/deploy/k3s/config/config.yaml index e2f37bc..b172eeb 100644 --- a/deploy/k3s/config/config.yaml +++ b/deploy/k3s/config/config.yaml @@ -129,8 +129,8 @@ llm: provider: ark base_url: https://ark.cn-beijing.volces.com/api/v3 api_key: "7fb6c66b-129c-4935-9617-709236c4205a" - model: doubao-seed-2-0-lite-260215 - knowledge_url_model: doubao-seed-2-0-mini-260215 + model: doubao-seed-2-0-pro-260215 # doubao-seed-2-0-lite-260215 # + knowledge_url_model: doubao-seed-2-0-mini-260215 #doubao-seed-2-0-mini-260215 timeout: 2m max_output_tokens: 16000 temperature: 0.7 @@ -172,3 +172,22 @@ generation: task_recovery_batch_size: 100 task_queued_stale_after: 2m task_max_attempts: 3 + +browser_fetch: + enabled: true + provider: lightpanda + service_url: http://browser-fetch:8082 + queue_size: 64 + worker_concurrency: 2 + request_timeout: 45s + cache_ttl: 6h + allowed_domains: + - zhuanlan.zhihu.com + - zhihu.com + trigger_status: + - 403 + - 429 + - 500 + - 502 + - 503 + - 504 diff --git a/deploy/k3s/kustomization.yaml b/deploy/k3s/kustomization.yaml index 89c3971..9f1e28d 100644 --- a/deploy/k3s/kustomization.yaml +++ b/deploy/k3s/kustomization.yaml @@ -37,6 +37,9 @@ images: - name: geo-rankly/tenant-api newName: geo-rankly/tenant-api newTag: latest + - name: geo-rankly/browser-fetch + newName: geo-rankly/browser-fetch + newTag: latest - name: geo-rankly/worker-generate newName: geo-rankly/worker-generate newTag: latest diff --git a/deploy/k3s/secrets.yaml b/deploy/k3s/secrets.yaml index bb47f86..a65df7f 100644 --- a/deploy/k3s/secrets.yaml +++ b/deploy/k3s/secrets.yaml @@ -12,6 +12,7 @@ stringData: OPS_JWT_SECRET: geo-ops-change-me-in-production OPS_DEFAULT_ADMIN_PASSWORD: Aa123456 SCHEDULER_INTERNAL_METRICS_TOKEN: geo-metrics-token + BROWSER_FETCH_TOKEN: geo-browser-fetch-change-me config.local.yaml: | database: password: ${POSTGRES_PASSWORD:geo_dev} @@ -31,3 +32,6 @@ stringData: scheduler: internal_metrics_token: "geo-metrics-token" + + browser_fetch: + token: ${BROWSER_FETCH_TOKEN:geo-browser-fetch-change-me} diff --git a/deploy/scripts/gitea-registry-images.sh b/deploy/scripts/gitea-registry-images.sh index d6834aa..f05f15f 100644 --- a/deploy/scripts/gitea-registry-images.sh +++ b/deploy/scripts/gitea-registry-images.sh @@ -16,7 +16,7 @@ REGISTRY_PASSWORD="${REGISTRY_PASSWORD:-}" REGISTRY_PLAIN_HTTP="${REGISTRY_PLAIN_HTTP:-true}" REGISTRY_SCHEME="${REGISTRY_SCHEME:-}" IMAGE_TAG="${IMAGE_TAG:-}" -IMAGE_LIST="${IMAGE_LIST:-tenant-api ops-api worker-generate kol-assist-worker scheduler frontend ops-web migrate}" +IMAGE_LIST="${IMAGE_LIST:-tenant-api browser-fetch ops-api worker-generate kol-assist-worker scheduler frontend ops-web migrate}" DELETE_OLD_IMAGE_TAGS="${DELETE_OLD_IMAGE_TAGS:-true}" PUSH_LATEST="${PUSH_LATEST:-true}" FORCE_BUILD="${FORCE_BUILD:-false}" @@ -257,6 +257,15 @@ build_image() { -f "${REPO_ROOT}/server/Dockerfile" \ "${REPO_ROOT}/server" ;; + browser-fetch) + docker build "${platform_args[@]}" \ + --target browser-fetch \ + --build-arg "SERVICE=browser-fetch" \ + --tag "${repo}:${IMAGE_TAG}" \ + --tag "${repo}:latest" \ + -f "${REPO_ROOT}/server/Dockerfile" \ + "${REPO_ROOT}/server" + ;; tenant-api|worker-generate|kol-assist-worker|scheduler|ops-api) docker build "${platform_args[@]}" \ --target runtime \ diff --git a/deploy/scripts/package.sh b/deploy/scripts/package.sh index f42f4d6..5d334a7 100755 --- a/deploy/scripts/package.sh +++ b/deploy/scripts/package.sh @@ -59,7 +59,7 @@ cd "${REPO_ROOT}" if [[ -n "${SOURCE_IMAGE_REGISTRY}" ]]; then echo " pulling from ${SOURCE_IMAGE_REGISTRY}" - for svc in tenant-api worker-generate kol-assist-worker ops-api scheduler migrate frontend ops-web; do + for svc in tenant-api browser-fetch worker-generate kol-assist-worker ops-api scheduler migrate frontend ops-web; do echo " pulling ${SOURCE_IMAGE_REGISTRY}/${svc}:${IMAGE_TAG}" docker pull --platform "${PLATFORM}" "${SOURCE_IMAGE_REGISTRY}/${svc}:${IMAGE_TAG}" docker tag "${SOURCE_IMAGE_REGISTRY}/${svc}:${IMAGE_TAG}" "${LOCAL_PREFIX}/${svc}:${IMAGE_TAG}" @@ -81,6 +81,7 @@ else } _build runtime tenant-api --build-arg SERVICE=tenant-api + _build browser-fetch browser-fetch --build-arg SERVICE=browser-fetch _build runtime worker-generate --build-arg SERVICE=worker-generate _build runtime kol-assist-worker --build-arg SERVICE=kol-assist-worker _build runtime ops-api --build-arg SERVICE=ops-api @@ -127,6 +128,7 @@ mkdir -p "${WORK_DIR}" ALL_IMAGES=( "${LOCAL_PREFIX}/tenant-api:${IMAGE_TAG}" + "${LOCAL_PREFIX}/browser-fetch:${IMAGE_TAG}" "${LOCAL_PREFIX}/worker-generate:${IMAGE_TAG}" "${LOCAL_PREFIX}/kol-assist-worker:${IMAGE_TAG}" "${LOCAL_PREFIX}/scheduler:${IMAGE_TAG}" diff --git a/server/Dockerfile b/server/Dockerfile index be1470d..b30c62d 100644 --- a/server/Dockerfile +++ b/server/Dockerfile @@ -23,6 +23,12 @@ RUN --mount=type=cache,target=/go/pkg/mod \ go build -trimpath -ldflags="-s -w" \ -o /bin/service ./cmd/${SERVICE} +RUN --mount=type=cache,target=/go/pkg/mod \ + --mount=type=cache,target=/root/.cache/go-build \ + CGO_ENABLED=0 GOOS="${TARGETOS:-$(go env GOOS)}" GOARCH="${TARGETARCH:-$(go env GOARCH)}" \ + go build -trimpath -ldflags="-s -w" \ + -o /bin/browser-fetch ./cmd/browser-fetch + RUN --mount=type=cache,target=/go/pkg/mod \ --mount=type=cache,target=/root/.cache/go-build \ CGO_ENABLED=1 GOOS="${TARGETOS:-$(go env GOOS)}" GOARCH="${TARGETARCH:-$(go env GOARCH)}" \ @@ -53,3 +59,13 @@ COPY configs/ /app/configs/ EXPOSE 8080 8081 ENTRYPOINT ["/app/service"] + +# ─── Stage 4: Browser fetch runtime (bundles Lightpanda) ────────────────────── +FROM lightpanda/browser:nightly AS browser-fetch + +WORKDIR /app + +COPY --from=builder /bin/browser-fetch /app/service + +EXPOSE 8082 +ENTRYPOINT ["/app/service"] diff --git a/server/cmd/browser-fetch/main.go b/server/cmd/browser-fetch/main.go new file mode 100644 index 0000000..3f55227 --- /dev/null +++ b/server/cmd/browser-fetch/main.go @@ -0,0 +1,877 @@ +package main + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net" + "net/http" + "net/url" + "os" + "os/exec" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" +) + +const ( + defaultPort = 8082 + defaultQueueSize = 128 + defaultWorkerConcurrency = 2 + defaultPerDomainConcurrency = 1 + defaultRequestTimeout = 45 * time.Second + defaultCacheTTL = 6 * time.Hour + defaultBreakerThreshold = 5 + defaultBreakerCooldown = 2 * time.Minute +) + +type fetchConfig struct { + Port int + Token string + QueueSize int + WorkerConcurrency int + PerDomainConcurrency int + RequestTimeout time.Duration + CacheTTL time.Duration + AllowedDomains []string + LightpandaDriver string + LightpandaBin string + LightpandaArgs []string + LightpandaFetchArgs []string + WorkerMaxRequests int + BreakerThreshold int + BreakerCooldown time.Duration +} + +type fetchService struct { + cfg fetchConfig + jobs chan fetchJob + cache *memoryCache + limiters *domainLimiters + breakers *domainBreakers + metrics *fetchMetrics + shuttingDown atomic.Bool +} + +type fetchJob struct { + rawURL string + timeout time.Duration + host string + release func() + resp chan fetchJobResult +} + +type fetchJobResult struct { + result fetchResponse + err error +} + +type fetchRequest struct { + URL string `json:"url"` + TimeoutMS int64 `json:"timeout_ms,omitempty"` + Options map[string]any `json:"options,omitempty"` +} + +type fetchResponse struct { + URL string `json:"url,omitempty"` + Title string `json:"title,omitempty"` + Text string `json:"text,omitempty"` + Markdown string `json:"markdown,omitempty"` +} + +type memoryCacheEntry struct { + result fetchResponse + expiresAt time.Time +} + +type memoryCache struct { + mu sync.Mutex + entries map[string]memoryCacheEntry +} + +type domainLimiters struct { + mu sync.Mutex + capacity int + items map[string]chan struct{} +} + +type domainBreakerState struct { + failures int + openedAt time.Time +} + +type domainBreakers struct { + mu sync.Mutex + threshold int + cooldown time.Duration + items map[string]domainBreakerState +} + +type fetchMetrics struct { + accepted atomic.Int64 + cacheHits atomic.Int64 + queueFull atomic.Int64 + domainLimited atomic.Int64 + authFailed atomic.Int64 + success atomic.Int64 + failed atomic.Int64 + breakerOpen atomic.Int64 +} + +type lightpandaWorker struct { + id int + cfg fetchConfig + client *mcpClient + handled int +} + +type mcpClient struct { + cfg fetchConfig + cmd *exec.Cmd + stdin io.WriteCloser + stdout *bufio.Scanner + stderr bytes.Buffer + nextID int64 + started bool +} + +type mcpResponse struct { + JSONRPC string `json:"jsonrpc"` + ID int64 `json:"id,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *mcpError `json:"error,omitempty"` +} + +type mcpError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +type mcpToolResult struct { + Content []mcpContentItem `json:"content"` + IsError bool `json:"isError"` +} + +type mcpContentItem struct { + Type string `json:"type"` + Text string `json:"text"` +} + +func main() { + cfg := loadFetchConfig() + if len(os.Args) > 1 && os.Args[1] == "-healthcheck" { + runHealthcheck(cfg) + return + } + service := newFetchService(cfg) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + service.Start(ctx) + + mux := http.NewServeMux() + mux.HandleFunc("/health/live", service.handleLive) + mux.HandleFunc("/health/ready", service.handleReady) + mux.HandleFunc("/metrics", service.handleMetrics) + mux.HandleFunc("/fetch", service.handleFetch) + + server := &http.Server{ + Addr: fmt.Sprintf(":%d", cfg.Port), + Handler: mux, + ReadHeaderTimeout: 5 * time.Second, + } + + log.Printf("browser-fetch listening on :%d workers=%d queue=%d domains=%v", cfg.Port, cfg.WorkerConcurrency, cfg.QueueSize, cfg.AllowedDomains) + if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Fatalf("browser-fetch server stopped: %v", err) + } +} + +func runHealthcheck(cfg fetchConfig) { + client := &http.Client{Timeout: 2 * time.Second} + resp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%d/health/ready", cfg.Port)) + if err != nil { + log.Fatalf("healthcheck failed: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode >= http.StatusBadRequest { + log.Fatalf("healthcheck failed: status=%d", resp.StatusCode) + } +} + +func loadFetchConfig() fetchConfig { + cfg := fetchConfig{ + Port: envInt("BROWSER_FETCH_PORT", defaultPort), + Token: strings.TrimSpace(os.Getenv("BROWSER_FETCH_TOKEN")), + QueueSize: envInt("BROWSER_FETCH_QUEUE_SIZE", defaultQueueSize), + WorkerConcurrency: envInt("BROWSER_FETCH_WORKER_CONCURRENCY", defaultWorkerConcurrency), + PerDomainConcurrency: envInt("BROWSER_FETCH_PER_DOMAIN_CONCURRENCY", defaultPerDomainConcurrency), + RequestTimeout: envDuration("BROWSER_FETCH_REQUEST_TIMEOUT", defaultRequestTimeout), + CacheTTL: envDuration("BROWSER_FETCH_CACHE_TTL", defaultCacheTTL), + AllowedDomains: splitCSV(os.Getenv("BROWSER_FETCH_ALLOWED_DOMAINS")), + LightpandaDriver: strings.ToLower(strings.TrimSpace(envString("LIGHTPANDA_DRIVER", "fetch"))), + LightpandaBin: strings.TrimSpace(envString("LIGHTPANDA_BIN", "lightpanda")), + LightpandaArgs: splitCommandArgs(envString("LIGHTPANDA_ARGS", "mcp")), + LightpandaFetchArgs: splitCommandArgs(envString("LIGHTPANDA_FETCH_ARGS", "--dump markdown")), + WorkerMaxRequests: envInt("LIGHTPANDA_WORKER_MAX_REQUESTS", 100), + BreakerThreshold: envInt("BROWSER_FETCH_BREAKER_THRESHOLD", defaultBreakerThreshold), + BreakerCooldown: envDuration("BROWSER_FETCH_BREAKER_COOLDOWN", defaultBreakerCooldown), + } + if cfg.QueueSize <= 0 { + cfg.QueueSize = defaultQueueSize + } + if cfg.WorkerConcurrency <= 0 { + cfg.WorkerConcurrency = defaultWorkerConcurrency + } + if cfg.PerDomainConcurrency <= 0 { + cfg.PerDomainConcurrency = defaultPerDomainConcurrency + } + if cfg.RequestTimeout <= 0 { + cfg.RequestTimeout = defaultRequestTimeout + } + if cfg.CacheTTL < 0 { + cfg.CacheTTL = 0 + } + if cfg.BreakerThreshold <= 0 { + cfg.BreakerThreshold = defaultBreakerThreshold + } + if cfg.BreakerCooldown <= 0 { + cfg.BreakerCooldown = defaultBreakerCooldown + } + if cfg.LightpandaDriver == "" { + cfg.LightpandaDriver = "fetch" + } + return cfg +} + +func newFetchService(cfg fetchConfig) *fetchService { + return &fetchService{ + cfg: cfg, + jobs: make(chan fetchJob, cfg.QueueSize), + cache: &memoryCache{entries: make(map[string]memoryCacheEntry)}, + limiters: &domainLimiters{capacity: cfg.PerDomainConcurrency, items: make(map[string]chan struct{})}, + breakers: &domainBreakers{threshold: cfg.BreakerThreshold, cooldown: cfg.BreakerCooldown, items: make(map[string]domainBreakerState)}, + metrics: &fetchMetrics{}, + } +} + +func (s *fetchService) Start(ctx context.Context) { + for i := 0; i < s.cfg.WorkerConcurrency; i++ { + worker := &lightpandaWorker{id: i, cfg: s.cfg} + go worker.Run(ctx, s.jobs, s.metrics, s.breakers) + } +} + +func (s *fetchService) handleLive(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "alive"}) +} + +func (s *fetchService) handleReady(w http.ResponseWriter, _ *http.Request) { + if s.cfg.Token == "" { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "BROWSER_FETCH_TOKEN is required"}) + return + } + if len(s.cfg.AllowedDomains) == 0 { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "BROWSER_FETCH_ALLOWED_DOMAINS is required"}) + return + } + writeJSON(w, http.StatusOK, map[string]string{"status": "ready"}) +} + +func (s *fetchService) handleMetrics(w http.ResponseWriter, r *http.Request) { + if !s.authorized(r) { + s.metrics.authFailed.Add(1) + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = fmt.Fprintf(w, "browser_fetch_requests_accepted_total %d\n", s.metrics.accepted.Load()) + _, _ = fmt.Fprintf(w, "browser_fetch_cache_hits_total %d\n", s.metrics.cacheHits.Load()) + _, _ = fmt.Fprintf(w, "browser_fetch_queue_full_total %d\n", s.metrics.queueFull.Load()) + _, _ = fmt.Fprintf(w, "browser_fetch_domain_limited_total %d\n", s.metrics.domainLimited.Load()) + _, _ = fmt.Fprintf(w, "browser_fetch_auth_failed_total %d\n", s.metrics.authFailed.Load()) + _, _ = fmt.Fprintf(w, "browser_fetch_success_total %d\n", s.metrics.success.Load()) + _, _ = fmt.Fprintf(w, "browser_fetch_failed_total %d\n", s.metrics.failed.Load()) + _, _ = fmt.Fprintf(w, "browser_fetch_breaker_open_total %d\n", s.metrics.breakerOpen.Load()) + _, _ = fmt.Fprintf(w, "browser_fetch_queue_depth %d\n", len(s.jobs)) +} + +func (s *fetchService) handleFetch(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"}) + return + } + if !s.authorized(r) { + s.metrics.authFailed.Add(1) + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + + var req fetchRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json"}) + return + } + rawURL := strings.TrimSpace(req.URL) + host, err := validateFetchURL(rawURL, s.cfg.AllowedDomains) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()}) + return + } + if s.breakers.IsOpen(host) { + s.metrics.breakerOpen.Add(1) + writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "domain circuit breaker is open"}) + return + } + + cacheKey := fetchCacheKey(rawURL) + if s.cfg.CacheTTL > 0 { + if cached, ok := s.cache.Get(cacheKey); ok { + s.metrics.cacheHits.Add(1) + writeJSON(w, http.StatusOK, cached) + return + } + } + + release, ok := s.limiters.TryAcquire(host) + if !ok { + s.metrics.domainLimited.Add(1) + writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "domain concurrency limit reached"}) + return + } + + timeout := s.cfg.RequestTimeout + if req.TimeoutMS > 0 { + timeout = time.Duration(req.TimeoutMS) * time.Millisecond + if timeout > s.cfg.RequestTimeout { + timeout = s.cfg.RequestTimeout + } + } + job := fetchJob{ + rawURL: rawURL, + timeout: timeout, + host: host, + release: release, + resp: make(chan fetchJobResult, 1), + } + + select { + case s.jobs <- job: + s.metrics.accepted.Add(1) + default: + release() + s.metrics.queueFull.Add(1) + writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "browser fetch queue is full"}) + return + } + + select { + case result := <-job.resp: + if result.err != nil { + writeJSON(w, http.StatusBadGateway, map[string]string{"error": result.err.Error()}) + return + } + if s.cfg.CacheTTL > 0 { + s.cache.Set(cacheKey, result.result, s.cfg.CacheTTL) + } + writeJSON(w, http.StatusOK, result.result) + case <-r.Context().Done(): + writeJSON(w, http.StatusGatewayTimeout, map[string]string{"error": "request cancelled"}) + } +} + +func (s *fetchService) authorized(r *http.Request) bool { + expected := s.cfg.Token + if expected == "" { + return false + } + value := strings.TrimSpace(r.Header.Get("Authorization")) + value = strings.TrimPrefix(value, "Bearer ") + if value == "" { + value = strings.TrimSpace(r.URL.Query().Get("token")) + } + return subtle.ConstantTimeCompare([]byte(value), []byte(expected)) == 1 +} + +func (w *lightpandaWorker) Run(ctx context.Context, jobs <-chan fetchJob, metrics *fetchMetrics, breakers *domainBreakers) { + defer w.closeClient() + for { + select { + case <-ctx.Done(): + return + case job := <-jobs: + result, err := w.Fetch(ctx, job.rawURL, job.timeout) + job.release() + if err != nil { + metrics.failed.Add(1) + breakers.RecordFailure(job.host) + } else { + metrics.success.Add(1) + breakers.RecordSuccess(job.host) + } + job.resp <- fetchJobResult{result: result, err: err} + } + } +} + +func (w *lightpandaWorker) Fetch(parent context.Context, rawURL string, timeout time.Duration) (fetchResponse, error) { + if timeout <= 0 { + timeout = w.cfg.RequestTimeout + } + ctx, cancel := context.WithTimeout(parent, timeout) + defer cancel() + + var markdown string + var err error + if w.cfg.LightpandaDriver == "mcp" { + markdown, err = w.fetchViaMCP(ctx, rawURL) + if err != nil { + w.closeClient() + return fetchResponse{}, err + } + } else { + markdown, err = w.fetchViaCommand(ctx, rawURL) + if err != nil { + return fetchResponse{}, err + } + } + w.handled++ + + markdown = strings.TrimSpace(markdown) + text := normalizeText(markdown) + if text == "" { + return fetchResponse{}, errors.New("lightpanda returned empty markdown") + } + return fetchResponse{ + URL: rawURL, + Title: extractMarkdownTitle(markdown), + Text: text, + Markdown: markdown, + }, nil +} + +func (w *lightpandaWorker) fetchViaCommand(ctx context.Context, rawURL string) (string, error) { + args := append([]string{"fetch"}, w.cfg.LightpandaFetchArgs...) + args = append(args, rawURL) + cmd := exec.CommandContext(ctx, w.cfg.LightpandaBin, args...) + var stderr bytes.Buffer + cmd.Stderr = &stderr + out, err := cmd.Output() + if err != nil { + msg := strings.TrimSpace(stderr.String()) + if msg != "" { + return "", fmt.Errorf("lightpanda fetch failed: %w: %s", err, truncateString(msg, 2048)) + } + return "", fmt.Errorf("lightpanda fetch failed: %w", err) + } + return string(out), nil +} + +func (w *lightpandaWorker) fetchViaMCP(ctx context.Context, rawURL string) (string, error) { + if w.client == nil || (w.cfg.WorkerMaxRequests > 0 && w.handled >= w.cfg.WorkerMaxRequests) { + w.closeClient() + w.client = newMCPClient(w.cfg) + w.handled = 0 + } + return w.client.Markdown(ctx, rawURL) +} + +func (w *lightpandaWorker) closeClient() { + if w.client != nil { + w.client.Close() + w.client = nil + } +} + +func newMCPClient(cfg fetchConfig) *mcpClient { + return &mcpClient{cfg: cfg} +} + +func (c *mcpClient) Markdown(ctx context.Context, rawURL string) (string, error) { + if err := c.ensureStarted(ctx); err != nil { + return "", err + } + id := atomic.AddInt64(&c.nextID, 1) + req := map[string]any{ + "jsonrpc": "2.0", + "id": id, + "method": "tools/call", + "params": map[string]any{ + "name": "markdown", + "arguments": map[string]any{ + "url": rawURL, + }, + }, + } + resp, err := c.call(ctx, id, req) + if err != nil { + return "", err + } + var toolResult mcpToolResult + if err := json.Unmarshal(resp.Result, &toolResult); err != nil { + return "", fmt.Errorf("decode lightpanda markdown result: %w", err) + } + texts := make([]string, 0, len(toolResult.Content)) + for _, item := range toolResult.Content { + if item.Type == "text" && strings.TrimSpace(item.Text) != "" { + texts = append(texts, item.Text) + } + } + text := strings.TrimSpace(strings.Join(texts, "\n\n")) + if toolResult.IsError { + if text == "" { + text = "lightpanda tool returned error" + } + return "", errors.New(text) + } + return text, nil +} + +func (c *mcpClient) ensureStarted(ctx context.Context) error { + if c.started { + return nil + } + args := append([]string{}, c.cfg.LightpandaArgs...) + cmd := exec.Command(c.cfg.LightpandaBin, args...) + stdin, err := cmd.StdinPipe() + if err != nil { + return fmt.Errorf("open lightpanda stdin: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return fmt.Errorf("open lightpanda stdout: %w", err) + } + cmd.Stderr = &c.stderr + if err := cmd.Start(); err != nil { + return fmt.Errorf("start lightpanda mcp: %w", err) + } + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 64*1024), 32*1024*1024) + c.cmd = cmd + c.stdin = stdin + c.stdout = scanner + c.started = true + + id := atomic.AddInt64(&c.nextID, 1) + req := map[string]any{ + "jsonrpc": "2.0", + "id": id, + "method": "initialize", + "params": map[string]any{ + "protocolVersion": "2024-11-05", + "capabilities": map[string]any{}, + "clientInfo": map[string]string{ + "name": "geo-browser-fetch", + "version": "1.0.0", + }, + }, + } + if _, err := c.call(ctx, id, req); err != nil { + c.Close() + return fmt.Errorf("initialize lightpanda mcp: %w", err) + } + initialized := map[string]any{ + "jsonrpc": "2.0", + "method": "notifications/initialized", + "params": map[string]any{}, + } + return c.writeJSON(initialized) +} + +func (c *mcpClient) call(ctx context.Context, id int64, payload map[string]any) (mcpResponse, error) { + if err := c.writeJSON(payload); err != nil { + return mcpResponse{}, err + } + type responseResult struct { + resp mcpResponse + err error + } + done := make(chan responseResult, 1) + go func() { + if c.stdout == nil { + done <- responseResult{err: errors.New("lightpanda mcp stdout is not open")} + return + } + for c.stdout.Scan() { + line := strings.TrimSpace(c.stdout.Text()) + if line == "" { + continue + } + var resp mcpResponse + if err := json.Unmarshal([]byte(line), &resp); err != nil { + continue + } + if resp.ID != id { + continue + } + if resp.Error != nil { + done <- responseResult{err: fmt.Errorf("lightpanda mcp error %d: %s", resp.Error.Code, resp.Error.Message)} + return + } + done <- responseResult{resp: resp} + return + } + err := c.stdout.Err() + if err == nil { + err = errors.New("lightpanda mcp stdout closed") + } + stderr := strings.TrimSpace(c.stderr.String()) + if stderr != "" { + err = fmt.Errorf("%w: %s", err, truncateString(stderr, 2048)) + } + done <- responseResult{err: err} + }() + + select { + case result := <-done: + return result.resp, result.err + case <-ctx.Done(): + return mcpResponse{}, ctx.Err() + } +} + +func (c *mcpClient) writeJSON(payload map[string]any) error { + data, err := json.Marshal(payload) + if err != nil { + return err + } + data = append(data, '\n') + _, err = c.stdin.Write(data) + return err +} + +func (c *mcpClient) Close() { + if c.stdin != nil { + _ = c.stdin.Close() + } + if c.cmd != nil && c.cmd.Process != nil { + _ = c.cmd.Process.Kill() + _, _ = c.cmd.Process.Wait() + } + c.started = false + c.cmd = nil + c.stdin = nil + c.stdout = nil + c.stderr.Reset() +} + +func (c *memoryCache) Get(key string) (fetchResponse, bool) { + now := time.Now() + c.mu.Lock() + defer c.mu.Unlock() + entry, ok := c.entries[key] + if !ok { + return fetchResponse{}, false + } + if now.After(entry.expiresAt) { + delete(c.entries, key) + return fetchResponse{}, false + } + return entry.result, true +} + +func (c *memoryCache) Set(key string, result fetchResponse, ttl time.Duration) { + if ttl <= 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + c.entries[key] = memoryCacheEntry{result: result, expiresAt: time.Now().Add(ttl)} +} + +func (l *domainLimiters) TryAcquire(host string) (func(), bool) { + l.mu.Lock() + ch := l.items[host] + if ch == nil { + ch = make(chan struct{}, l.capacity) + l.items[host] = ch + } + l.mu.Unlock() + select { + case ch <- struct{}{}: + return func() { <-ch }, true + default: + return func() {}, false + } +} + +func (b *domainBreakers) IsOpen(host string) bool { + b.mu.Lock() + defer b.mu.Unlock() + state := b.items[host] + if state.openedAt.IsZero() { + return false + } + if time.Since(state.openedAt) >= b.cooldown { + delete(b.items, host) + return false + } + return true +} + +func (b *domainBreakers) RecordSuccess(host string) { + b.mu.Lock() + defer b.mu.Unlock() + delete(b.items, host) +} + +func (b *domainBreakers) RecordFailure(host string) { + b.mu.Lock() + defer b.mu.Unlock() + state := b.items[host] + if !state.openedAt.IsZero() && time.Since(state.openedAt) < b.cooldown { + b.items[host] = state + return + } + state.failures++ + if state.failures >= b.threshold { + state.openedAt = time.Now() + } + b.items[host] = state +} + +func validateFetchURL(rawURL string, allowedDomains []string) (string, error) { + parsed, err := url.Parse(rawURL) + if err != nil || parsed.Scheme == "" || parsed.Host == "" { + return "", errors.New("invalid url") + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return "", errors.New("only http and https urls are allowed") + } + host := strings.ToLower(strings.TrimSpace(parsed.Hostname())) + if host == "" { + return "", errors.New("url host is required") + } + if ip := net.ParseIP(host); ip != nil { + return "", errors.New("ip literal urls are not allowed") + } + if !domainAllowed(host, allowedDomains) { + return "", errors.New("domain is not allowed") + } + return host, nil +} + +func domainAllowed(host string, allowedDomains []string) bool { + if len(allowedDomains) == 0 { + return false + } + for _, domain := range allowedDomains { + domain = strings.ToLower(strings.TrimPrefix(strings.TrimSpace(domain), ".")) + if domain == "" { + continue + } + if host == domain || strings.HasSuffix(host, "."+domain) { + return true + } + } + return false +} + +func fetchCacheKey(rawURL string) string { + normalized := strings.TrimSpace(rawURL) + if parsed, err := url.Parse(normalized); err == nil && parsed.Host != "" { + parsed.Fragment = "" + normalized = parsed.String() + } + sum := sha256.Sum256([]byte(normalized)) + return hex.EncodeToString(sum[:]) +} + +func extractMarkdownTitle(markdown string) string { + for _, line := range strings.Split(markdown, "\n") { + line = strings.TrimSpace(line) + if strings.HasPrefix(line, "# ") { + return strings.TrimSpace(strings.TrimPrefix(line, "# ")) + } + } + return "" +} + +func normalizeText(value string) string { + lines := strings.Split(value, "\n") + out := make([]string, 0, len(lines)) + for _, line := range lines { + line = strings.TrimSpace(line) + if line != "" { + out = append(out, line) + } + } + return strings.Join(out, "\n") +} + +func writeJSON(w http.ResponseWriter, status int, payload any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(payload) +} + +func splitCSV(value string) []string { + parts := strings.Split(value, ",") + out := make([]string, 0, len(parts)) + for _, part := range parts { + part = strings.TrimSpace(part) + if part != "" { + out = append(out, part) + } + } + return out +} + +func splitCommandArgs(value string) []string { + parts := strings.Fields(value) + if len(parts) == 0 { + return nil + } + return parts +} + +func envString(key, fallback string) string { + if value := strings.TrimSpace(os.Getenv(key)); value != "" { + return value + } + return fallback +} + +func envInt(key string, fallback int) int { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return parsed +} + +func envDuration(key string, fallback time.Duration) time.Duration { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := time.ParseDuration(value) + if err == nil { + return parsed + } + ms, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return time.Duration(ms) * time.Millisecond +} + +func truncateString(value string, max int) string { + if len(value) <= max { + return value + } + return value[:max] +} diff --git a/server/configs/config.local.yaml.example b/server/configs/config.local.yaml.example index 7534954..e407bd7 100644 --- a/server/configs/config.local.yaml.example +++ b/server/configs/config.local.yaml.example @@ -204,3 +204,23 @@ generation: worker_concurrency: 8 stream_enabled: false article_timeout: 8m + +browser_fetch: + enabled: false + provider: lightpanda + service_url: http://localhost:8082 + token: dev-browser-fetch-token + queue_size: 64 + worker_concurrency: 2 + request_timeout: 45s + cache_ttl: 6h + allowed_domains: + - zhuanlan.zhihu.com + - zhihu.com + trigger_status: + - 403 + - 429 + - 500 + - 502 + - 503 + - 504 diff --git a/server/configs/config.yaml b/server/configs/config.yaml index 5561e3a..f6695ae 100644 --- a/server/configs/config.yaml +++ b/server/configs/config.yaml @@ -216,4 +216,24 @@ generation: task_recovery_batch_size: 100 task_queued_stale_after: 2m task_max_attempts: 3 + +browser_fetch: + enabled: false + provider: lightpanda + service_url: http://localhost:8082 + token: dev-browser-fetch-token + queue_size: 64 + worker_concurrency: 2 + request_timeout: 45s + cache_ttl: 6h + allowed_domains: + - zhuanlan.zhihu.com + - zhihu.com + trigger_status: + - 403 + - 429 + - 500 + - 502 + - 503 + - 504 diff --git a/server/internal/shared/config/config.go b/server/internal/shared/config/config.go index 0101852..3c2a600 100644 --- a/server/internal/shared/config/config.go +++ b/server/internal/shared/config/config.go @@ -38,6 +38,7 @@ type Config struct { Compliance ComplianceConfig `mapstructure:"compliance"` Retrieval RetrievalConfig `mapstructure:"retrieval"` Generation GenerationConfig `mapstructure:"generation"` + BrowserFetch BrowserFetchConfig `mapstructure:"browser_fetch"` } type ServerConfig struct { @@ -348,6 +349,19 @@ type GenerationConfig struct { TaskStateCheckLookback time.Duration `mapstructure:"task_state_check_lookback"` } +type BrowserFetchConfig struct { + Enabled bool `mapstructure:"enabled"` + Provider string `mapstructure:"provider"` + ServiceURL string `mapstructure:"service_url"` + Token string `mapstructure:"token"` + QueueSize int `mapstructure:"queue_size"` + WorkerConcurrency int `mapstructure:"worker_concurrency"` + RequestTimeout time.Duration `mapstructure:"request_timeout"` + CacheTTL time.Duration `mapstructure:"cache_ttl"` + AllowedDomains []string `mapstructure:"allowed_domains"` + TriggerStatus []int `mapstructure:"trigger_status"` +} + func Load(configPath string) (*Config, error) { cfg, _, err := loadWithFiles(configPath) return cfg, err @@ -382,6 +396,7 @@ func loadWithFiles(configPath string) (*Config, []string, error) { normalizeBrandLibraryConfig(&cfg.BrandLibrary) NormalizeComplianceConfig(&cfg.Compliance) NormalizeGenerationConfig(&cfg.Generation) + NormalizeBrowserFetchConfig(&cfg.BrowserFetch) files := []string{configFile} if localConfigFile != "" { @@ -511,6 +526,42 @@ func compactStringSlice(values []string) []string { return out } +func compactIntSlice(values []int) []int { + if values == nil { + return nil + } + out := make([]int, 0, len(values)) + seen := make(map[int]struct{}, len(values)) + for _, value := range values { + if value <= 0 { + continue + } + if _, exists := seen[value]; exists { + continue + } + seen[value] = struct{}{} + out = append(out, value) + } + return out +} + +func parseIntList(value string) []int { + parts := strings.Split(value, ",") + out := make([]int, 0, len(parts)) + for _, part := range parts { + part = strings.TrimSpace(part) + if part == "" { + continue + } + n, err := strconv.Atoi(part) + if err != nil { + continue + } + out = append(out, n) + } + return out +} + func normalizeMonitoringDispatchConfig(cfg *MonitoringDispatchConfig) { if cfg == nil { return @@ -643,6 +694,35 @@ func NormalizeGenerationConfig(cfg *GenerationConfig) { } } +func NormalizeBrowserFetchConfig(cfg *BrowserFetchConfig) { + if cfg == nil { + return + } + cfg.Provider = strings.ToLower(strings.TrimSpace(cfg.Provider)) + if cfg.Provider == "" { + cfg.Provider = "lightpanda" + } + cfg.ServiceURL = strings.TrimRight(strings.TrimSpace(cfg.ServiceURL), "/") + cfg.Token = strings.TrimSpace(cfg.Token) + if cfg.QueueSize <= 0 { + cfg.QueueSize = 64 + } + if cfg.WorkerConcurrency <= 0 { + cfg.WorkerConcurrency = 1 + } + if cfg.RequestTimeout <= 0 { + cfg.RequestTimeout = 45 * time.Second + } + if cfg.CacheTTL <= 0 { + cfg.CacheTTL = 6 * time.Hour + } + cfg.AllowedDomains = compactStringSlice(cfg.AllowedDomains) + cfg.TriggerStatus = compactIntSlice(cfg.TriggerStatus) + if len(cfg.TriggerStatus) == 0 { + cfg.TriggerStatus = []int{403, 429, 500, 502, 503, 504} + } +} + func NormalizeComplianceConfig(cfg *ComplianceConfig) { if cfg == nil { return @@ -759,6 +839,36 @@ func applyEnvOverrides(cfg *Config) { if d, ok := lookupDurationEnv("GENERATION_TASK_STATE_CHECK_LOOKBACK"); ok { cfg.Generation.TaskStateCheckLookback = d } + if enabled, ok := lookupBoolEnv("BROWSER_FETCH_ENABLED"); ok { + cfg.BrowserFetch.Enabled = enabled + } + if provider, ok := lookupNonEmptyEnv("BROWSER_FETCH_PROVIDER"); ok { + cfg.BrowserFetch.Provider = provider + } + if serviceURL, ok := lookupFirstNonEmptyEnv("BROWSER_FETCH_SERVICE_URL", "LIGHTPANDA_BROWSER_FETCH_URL"); ok { + cfg.BrowserFetch.ServiceURL = serviceURL + } + if token, ok := lookupNonEmptyEnv("BROWSER_FETCH_TOKEN"); ok { + cfg.BrowserFetch.Token = token + } + if n, ok := lookupIntEnv("BROWSER_FETCH_QUEUE_SIZE"); ok { + cfg.BrowserFetch.QueueSize = n + } + if n, ok := lookupIntEnv("BROWSER_FETCH_WORKER_CONCURRENCY"); ok { + cfg.BrowserFetch.WorkerConcurrency = n + } + if d, ok := lookupDurationEnv("BROWSER_FETCH_REQUEST_TIMEOUT"); ok { + cfg.BrowserFetch.RequestTimeout = d + } + if d, ok := lookupDurationEnv("BROWSER_FETCH_CACHE_TTL"); ok { + cfg.BrowserFetch.CacheTTL = d + } + if domains, ok := lookupNonEmptyEnv("BROWSER_FETCH_ALLOWED_DOMAINS"); ok { + cfg.BrowserFetch.AllowedDomains = strings.Split(domains, ",") + } + if statuses, ok := lookupNonEmptyEnv("BROWSER_FETCH_TRIGGER_STATUS"); ok { + cfg.BrowserFetch.TriggerStatus = parseIntList(statuses) + } if apiKey, ok := lookupNonEmptyEnv("QDRANT_API_KEY"); ok { cfg.Qdrant.APIKey = apiKey } diff --git a/server/internal/shared/config/config_test.go b/server/internal/shared/config/config_test.go index 5f54f29..34b407d 100644 --- a/server/internal/shared/config/config_test.go +++ b/server/internal/shared/config/config_test.go @@ -557,6 +557,93 @@ generation: {} } } +func TestLoadAppliesBrowserFetchDefaultsAndOverrides(t *testing.T) { + configPath := writeTestConfig(t, ` +browser_fetch: + enabled: true + service_url: " http://browser-fetch:8082/ " + token: " token " + allowed_domains: + - " zhuanlan.zhihu.com " + - "" + trigger_status: + - 403 + - 403 + - 500 +`) + + cfg, err := Load(configPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + + if !cfg.BrowserFetch.Enabled { + t.Fatalf("expected browser fetch enabled") + } + if cfg.BrowserFetch.Provider != "lightpanda" { + t.Fatalf("expected default provider lightpanda, got %q", cfg.BrowserFetch.Provider) + } + if cfg.BrowserFetch.ServiceURL != "http://browser-fetch:8082" { + t.Fatalf("expected normalized service url, got %q", cfg.BrowserFetch.ServiceURL) + } + if cfg.BrowserFetch.Token != "token" { + t.Fatalf("expected normalized token, got %q", cfg.BrowserFetch.Token) + } + if cfg.BrowserFetch.QueueSize != 64 || + cfg.BrowserFetch.WorkerConcurrency != 1 || + cfg.BrowserFetch.RequestTimeout != 45*time.Second || + cfg.BrowserFetch.CacheTTL != 6*time.Hour { + t.Fatalf("unexpected browser fetch defaults: %#v", cfg.BrowserFetch) + } + if len(cfg.BrowserFetch.AllowedDomains) != 1 || cfg.BrowserFetch.AllowedDomains[0] != "zhuanlan.zhihu.com" { + t.Fatalf("unexpected allowed domains: %#v", cfg.BrowserFetch.AllowedDomains) + } + if len(cfg.BrowserFetch.TriggerStatus) != 2 || + cfg.BrowserFetch.TriggerStatus[0] != 403 || + cfg.BrowserFetch.TriggerStatus[1] != 500 { + t.Fatalf("unexpected trigger status: %#v", cfg.BrowserFetch.TriggerStatus) + } +} + +func TestLoadAppliesBrowserFetchEnvOverrides(t *testing.T) { + t.Setenv("BROWSER_FETCH_ENABLED", "true") + t.Setenv("BROWSER_FETCH_PROVIDER", "lightpanda_service") + t.Setenv("BROWSER_FETCH_SERVICE_URL", "http://env-browser-fetch:8082/") + t.Setenv("BROWSER_FETCH_TOKEN", "env-token") + t.Setenv("BROWSER_FETCH_QUEUE_SIZE", "11") + t.Setenv("BROWSER_FETCH_WORKER_CONCURRENCY", "3") + t.Setenv("BROWSER_FETCH_REQUEST_TIMEOUT", "12s") + t.Setenv("BROWSER_FETCH_CACHE_TTL", "30m") + t.Setenv("BROWSER_FETCH_ALLOWED_DOMAINS", "zhihu.com,example.com") + t.Setenv("BROWSER_FETCH_TRIGGER_STATUS", "403,429,broken,503") + + configPath := writeTestConfig(t, ` +browser_fetch: {} +`) + + cfg, err := Load(configPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + + if !cfg.BrowserFetch.Enabled || + cfg.BrowserFetch.Provider != "lightpanda_service" || + cfg.BrowserFetch.ServiceURL != "http://env-browser-fetch:8082" || + cfg.BrowserFetch.Token != "env-token" || + cfg.BrowserFetch.QueueSize != 11 || + cfg.BrowserFetch.WorkerConcurrency != 3 || + cfg.BrowserFetch.RequestTimeout != 12*time.Second || + cfg.BrowserFetch.CacheTTL != 30*time.Minute { + t.Fatalf("unexpected browser fetch env config: %#v", cfg.BrowserFetch) + } + if got := cfg.BrowserFetch.AllowedDomains; len(got) != 2 || got[0] != "zhihu.com" || got[1] != "example.com" { + t.Fatalf("unexpected env allowed domains: %#v", got) + } + if got := cfg.BrowserFetch.TriggerStatus; len(got) != 3 || got[0] != 403 || got[1] != 429 || got[2] != 503 { + t.Fatalf("unexpected env trigger status: %#v", got) + } +} + func TestStoreReloadsConfigAndKeepsPreviousOnInvalidYAML(t *testing.T) { configPath := writeTestConfig(t, ` jwt: diff --git a/server/internal/shared/config/reload.go b/server/internal/shared/config/reload.go index 8a6b85a..0709aca 100644 --- a/server/internal/shared/config/reload.go +++ b/server/internal/shared/config/reload.go @@ -80,6 +80,12 @@ func Diff(previous, current *Config) []FieldChange { if previous.Retrieval != current.Retrieval { addChange("retrieval", true) } + if previous.BrowserFetch.QueueSize != current.BrowserFetch.QueueSize || + previous.BrowserFetch.WorkerConcurrency != current.BrowserFetch.WorkerConcurrency { + addChange("browser_fetch.worker", false) + } else if !reflect.DeepEqual(previous.BrowserFetch, current.BrowserFetch) { + addChange("browser_fetch", true) + } if previous.Generation.QueueSize != current.Generation.QueueSize || previous.Generation.WorkerConcurrency != current.Generation.WorkerConcurrency { addChange("generation.worker", false) diff --git a/server/internal/tenant/app/browser_fetch_queue.go b/server/internal/tenant/app/browser_fetch_queue.go new file mode 100644 index 0000000..05d9565 --- /dev/null +++ b/server/internal/tenant/app/browser_fetch_queue.go @@ -0,0 +1,236 @@ +package app + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "time" + + "github.com/geo-platform/tenant-api/internal/shared/config" + "go.uber.org/zap" +) + +var errBrowserFetchQueueFull = errors.New("browser fetch queue is full") + +type browserFetchResult struct { + URL string `json:"url,omitempty"` + Title string `json:"title,omitempty"` + Text string `json:"text,omitempty"` + Markdown string `json:"markdown,omitempty"` +} + +type browserFetchRequest struct { + URL string `json:"url"` + TimeoutMS int64 `json:"timeout_ms,omitempty"` + Options map[string]interface{} `json:"options,omitempty"` + ctx context.Context `json:"-"` + cfg config.BrowserFetchConfig `json:"-"` + resp chan browserFetchRequestResult `json:"-"` +} + +type browserFetchRequestResult struct { + result browserFetchResult + err error +} + +type browserFetchQueue struct { + logger *zap.Logger + client *http.Client + jobs chan browserFetchRequest + cache *browserFetchMemoryCache +} + +type browserFetchCacheEntry struct { + result browserFetchResult + expiresAt time.Time +} + +type browserFetchMemoryCache struct { + mu sync.Mutex + entries map[string]browserFetchCacheEntry +} + +func (s *KnowledgeService) browserFetchQueue(cfg config.BrowserFetchConfig) *browserFetchQueue { + s.browserFetchMu.Lock() + defer s.browserFetchMu.Unlock() + if s.browserFetchQ != nil { + return s.browserFetchQ + } + + config.NormalizeBrowserFetchConfig(&cfg) + queue := &browserFetchQueue{ + logger: s.logger, + client: &http.Client{}, + jobs: make(chan browserFetchRequest, cfg.QueueSize), + cache: &browserFetchMemoryCache{ + entries: make(map[string]browserFetchCacheEntry), + }, + } + + for i := 0; i < cfg.WorkerConcurrency; i++ { + go queue.runWorker(i) + } + s.browserFetchQ = queue + return queue +} + +func (q *browserFetchQueue) Fetch(ctx context.Context, rawURL string, cfg config.BrowserFetchConfig) (browserFetchResult, error) { + if q == nil { + return browserFetchResult{}, errBrowserFetchNotAvailable + } + config.NormalizeBrowserFetchConfig(&cfg) + cacheKey := browserFetchCacheKey(rawURL) + if cfg.CacheTTL > 0 { + if result, ok := q.cache.Get(cacheKey); ok { + return result, nil + } + } + + request := browserFetchRequest{ + URL: rawURL, + TimeoutMS: int64(cfg.RequestTimeout / time.Millisecond), + Options: map[string]interface{}{ + "provider": cfg.Provider, + }, + ctx: ctx, + cfg: cfg, + resp: make(chan browserFetchRequestResult, 1), + } + + select { + case q.jobs <- request: + case <-ctx.Done(): + return browserFetchResult{}, ctx.Err() + default: + return browserFetchResult{}, errBrowserFetchQueueFull + } + + select { + case result := <-request.resp: + if result.err != nil { + return browserFetchResult{}, result.err + } + if cfg.CacheTTL > 0 { + q.cache.Set(cacheKey, result.result, cfg.CacheTTL) + } + return result.result, nil + case <-ctx.Done(): + return browserFetchResult{}, ctx.Err() + } +} + +func (q *browserFetchQueue) runWorker(workerID int) { + for request := range q.jobs { + result, err := q.execute(context.Background(), request) + if err != nil && q.logger != nil { + q.logger.Warn("browser fetch request failed", + zap.Int("worker_id", workerID), + zap.String("url", request.URL), + zap.Error(err), + ) + } + request.resp <- browserFetchRequestResult{result: result, err: err} + } +} + +func (q *browserFetchQueue) execute(parent context.Context, request browserFetchRequest) (browserFetchResult, error) { + if request.ctx != nil { + parent = request.ctx + } + timeout := time.Duration(request.TimeoutMS) * time.Millisecond + if timeout <= 0 { + timeout = 45 * time.Second + } + ctx, cancel := context.WithTimeout(parent, timeout) + defer cancel() + + return q.callService(ctx, request.URL, request.cfg) +} + +func (q *browserFetchQueue) callService(ctx context.Context, rawURL string, cfg config.BrowserFetchConfig) (browserFetchResult, error) { + config.NormalizeBrowserFetchConfig(&cfg) + endpoint := strings.TrimRight(cfg.ServiceURL, "/") + "/fetch" + payload := browserFetchRequest{ + URL: rawURL, + TimeoutMS: int64(cfg.RequestTimeout / time.Millisecond), + } + body, err := json.Marshal(payload) + if err != nil { + return browserFetchResult{}, fmt.Errorf("marshal browser fetch request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + return browserFetchResult{}, fmt.Errorf("create browser fetch request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + if cfg.Token != "" { + req.Header.Set("Authorization", "Bearer "+cfg.Token) + } + + resp, err := q.client.Do(req) + if err != nil { + return browserFetchResult{}, fmt.Errorf("execute browser fetch request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= http.StatusBadRequest { + data, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return browserFetchResult{}, fmt.Errorf("browser fetch service failed: status=%d body=%s", resp.StatusCode, strings.TrimSpace(string(data))) + } + + var result browserFetchResult + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return browserFetchResult{}, fmt.Errorf("decode browser fetch response: %w", err) + } + return result, nil +} + +func (c *browserFetchMemoryCache) Get(key string) (browserFetchResult, bool) { + if c == nil || key == "" { + return browserFetchResult{}, false + } + now := time.Now() + c.mu.Lock() + defer c.mu.Unlock() + entry, ok := c.entries[key] + if !ok { + return browserFetchResult{}, false + } + if now.After(entry.expiresAt) { + delete(c.entries, key) + return browserFetchResult{}, false + } + return entry.result, true +} + +func (c *browserFetchMemoryCache) Set(key string, result browserFetchResult, ttl time.Duration) { + if c == nil || key == "" || ttl <= 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + c.entries[key] = browserFetchCacheEntry{ + result: result, + expiresAt: time.Now().Add(ttl), + } +} + +func browserFetchCacheKey(rawURL string) string { + normalized := strings.TrimSpace(rawURL) + if parsed, err := url.Parse(normalized); err == nil && parsed.Host != "" { + parsed.Fragment = "" + normalized = parsed.String() + } + sum := sha256.Sum256([]byte(normalized)) + return hex.EncodeToString(sum[:]) +} diff --git a/server/internal/tenant/app/knowledge_service.go b/server/internal/tenant/app/knowledge_service.go index 11931fd..2b62e50 100644 --- a/server/internal/tenant/app/knowledge_service.go +++ b/server/internal/tenant/app/knowledge_service.go @@ -49,6 +49,8 @@ type KnowledgeService struct { jobs chan knowledgeParseJob cleanupMu sync.Mutex cleanupInFlight map[string]struct{} + browserFetchMu sync.Mutex + browserFetchQ *browserFetchQueue } type knowledgeParseJob struct { @@ -199,7 +201,7 @@ func (s *KnowledgeService) WithConfigProvider(provider runtimeConfigProvider) *K func (s *KnowledgeService) runtimeConfig() knowledgeRuntimeConfig { if s != nil && s.configProvider != nil { if cfg := s.configProvider.Current(); cfg != nil { - return knowledgeRuntimeFromConfig(cfg.Retrieval, cfg.LLM) + return knowledgeRuntimeFromFullConfig(cfg) } } if s == nil { diff --git a/server/internal/tenant/app/knowledge_url_parser.go b/server/internal/tenant/app/knowledge_url_parser.go index 4f9b1c9..3930731 100644 --- a/server/internal/tenant/app/knowledge_url_parser.go +++ b/server/internal/tenant/app/knowledge_url_parser.go @@ -4,12 +4,16 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" + "net/url" + "strconv" "strings" "time" + "github.com/geo-platform/tenant-api/internal/shared/config" "github.com/geo-platform/tenant-api/internal/shared/llm" "github.com/geo-platform/tenant-api/internal/tenant/prompts" "go.uber.org/zap" @@ -23,6 +27,37 @@ const ( defaultKnowledgeURLMarkdownOutputToken = 12000 ) +var errBrowserFetchNotAvailable = fmt.Errorf("browser fetch fallback is not available") + +type sourceFetchRestrictedError struct { + URL string + Status int + Cause error +} + +func (e sourceFetchRestrictedError) Error() string { + cause := "" + if e.Cause != nil { + cause = strings.TrimSpace(e.Cause.Error()) + } + if cause == "" { + cause = "source website requires browser rendering or blocks server-side fetch" + } + if e.Status > 0 { + return fmt.Sprintf("source website fetch restricted status=%d: %s", e.Status, cause) + } + return "source website fetch restricted: " + cause +} + +func (e sourceFetchRestrictedError) Unwrap() error { + return e.Cause +} + +func isSourceFetchRestrictedError(err error) bool { + var restricted sourceFetchRestrictedError + return errors.As(err, &restricted) +} + type arkToolExecuteRequest struct { ActionName string `json:"action_name"` ToolName string `json:"tool_name"` @@ -50,7 +85,20 @@ type arkWebDataItem struct { func (s *KnowledgeService) parseWebsiteKnowledge(ctx context.Context, rawURL string) (*knowledgeParsedContent, error) { item, err := s.executeLinkReader(ctx, rawURL) if err != nil { - return nil, err + if !s.shouldTryBrowserFetchFallback(rawURL, err) { + return nil, classifySourceFetchError(rawURL, err) + } + fallback, fallbackErr := s.parseWebsiteKnowledgeWithBrowserFetch(ctx, rawURL) + if fallbackErr == nil { + return fallback, nil + } + if s.logger != nil { + s.logger.Warn("knowledge browser fetch fallback failed", + zap.String("url", rawURL), + zap.Error(fallbackErr), + ) + } + return nil, classifySourceFetchError(rawURL, err) } text := normalizeKnowledgeText(item.Content) @@ -76,6 +124,65 @@ func (s *KnowledgeService) parseWebsiteKnowledge(ctx context.Context, rawURL str }, nil } +func (s *KnowledgeService) parseWebsiteKnowledgeWithBrowserFetch(ctx context.Context, rawURL string) (*knowledgeParsedContent, error) { + runtimeCfg := s.runtimeConfig() + browserCfg := runtimeCfg.BrowserFetch + config.NormalizeBrowserFetchConfig(&browserCfg) + if !browserCfg.Enabled { + return nil, errBrowserFetchNotAvailable + } + + provider := strings.ToLower(strings.TrimSpace(browserCfg.Provider)) + if provider != "lightpanda" && provider != "lightpanda_service" { + return nil, fmt.Errorf("unsupported browser fetch provider %q", browserCfg.Provider) + } + + if strings.TrimSpace(browserCfg.ServiceURL) == "" { + return nil, fmt.Errorf("%w: browser_fetch.service_url is empty", errBrowserFetchNotAvailable) + } + + callCtx := ctx + if browserCfg.RequestTimeout > 0 { + var cancel context.CancelFunc + callCtx, cancel = context.WithTimeout(ctx, browserCfg.RequestTimeout) + defer cancel() + } + + result, err := s.browserFetchQueue(browserCfg).Fetch(callCtx, rawURL, browserCfg) + if err != nil { + return nil, err + } + + text := normalizeKnowledgeText(result.Text) + if text == "" { + text = normalizeKnowledgeText(result.Markdown) + } + if text == "" { + return nil, fmt.Errorf("browser fetch returned empty content") + } + + title := strings.TrimSpace(result.Title) + markdown := strings.TrimSpace(result.Markdown) + if markdown == "" { + var err error + markdown, err = s.formatWebsiteMarkdown(ctx, title, text, rawURL) + if err != nil { + if s.logger != nil { + s.logger.Warn("knowledge browser fetch markdown formatting failed, fallback to local markdown", + zap.String("url", rawURL), + zap.Error(err), + ) + } + markdown = buildKnowledgeMarkdownFromText(title, text) + } + } + + return &knowledgeParsedContent{ + Text: text, + Markdown: markdown, + }, nil +} + func (s *KnowledgeService) executeLinkReader(ctx context.Context, rawURL string) (*arkWebDataItem, error) { runtimeCfg := s.runtimeConfig() baseURL := strings.TrimRight(strings.TrimSpace(runtimeCfg.ArkBaseURL), "/") @@ -135,6 +242,90 @@ func (s *KnowledgeService) executeLinkReader(ctx context.Context, rawURL string) return &item, nil } +func classifySourceFetchError(rawURL string, err error) error { + if err == nil { + return nil + } + status := extractHTTPStatusFromError(err) + if status == http.StatusForbidden || status == http.StatusTooManyRequests || status >= http.StatusInternalServerError { + return sourceFetchRestrictedError{URL: rawURL, Status: status, Cause: err} + } + return err +} + +func extractHTTPStatusFromError(err error) int { + if err == nil { + return 0 + } + message := err.Error() + idx := strings.Index(message, "status=") + if idx < 0 { + return 0 + } + value := message[idx+len("status="):] + end := 0 + for end < len(value) && value[end] >= '0' && value[end] <= '9' { + end++ + } + if end == 0 { + return 0 + } + status, parseErr := strconv.Atoi(value[:end]) + if parseErr != nil { + return 0 + } + return status +} + +func (s *KnowledgeService) shouldTryBrowserFetchFallback(rawURL string, err error) bool { + if s == nil { + return false + } + runtimeCfg := s.runtimeConfig() + browserCfg := runtimeCfg.BrowserFetch + config.NormalizeBrowserFetchConfig(&browserCfg) + if !browserCfg.Enabled { + return false + } + if !browserFetchDomainAllowed(rawURL, browserCfg.AllowedDomains) { + return false + } + status := extractHTTPStatusFromError(err) + if status <= 0 { + return false + } + for _, trigger := range browserCfg.TriggerStatus { + if status == trigger { + return true + } + } + return false +} + +func browserFetchDomainAllowed(rawURL string, allowedDomains []string) bool { + if len(allowedDomains) == 0 { + return false + } + parsed, err := url.Parse(rawURL) + if err != nil { + return false + } + host := strings.ToLower(strings.TrimSpace(parsed.Hostname())) + if host == "" { + return false + } + for _, domain := range allowedDomains { + domain = strings.ToLower(strings.TrimPrefix(strings.TrimSpace(domain), ".")) + if domain == "" { + continue + } + if host == domain || strings.HasSuffix(host, "."+domain) { + return true + } + } + return false +} + func (s *KnowledgeService) formatWebsiteMarkdown(ctx context.Context, title, content, rawURL string) (string, error) { content = strings.TrimSpace(content) if content == "" { diff --git a/server/internal/tenant/app/runtime_config.go b/server/internal/tenant/app/runtime_config.go index cd7c38b..13f28a8 100644 --- a/server/internal/tenant/app/runtime_config.go +++ b/server/internal/tenant/app/runtime_config.go @@ -26,6 +26,7 @@ type knowledgeRuntimeConfig struct { ArkBaseURL string ArkAPIKey string URLMarkdownModel string + BrowserFetch config.BrowserFetchConfig } type generationStreamDisabledConfigProvider struct { @@ -103,3 +104,14 @@ func knowledgeRuntimeFromConfig(retrievalCfg config.RetrievalConfig, llmCfg conf URLMarkdownModel: model, } } + +func knowledgeRuntimeFromFullConfig(cfg *config.Config) knowledgeRuntimeConfig { + if cfg == nil { + return knowledgeRuntimeFromConfig(config.RetrievalConfig{}, config.LLMConfig{}) + } + runtime := knowledgeRuntimeFromConfig(cfg.Retrieval, cfg.LLM) + browserFetch := cfg.BrowserFetch + config.NormalizeBrowserFetchConfig(&browserFetch) + runtime.BrowserFetch = browserFetch + return runtime +}