網頁抓取與嵌入向量流程：端到端構建可引用的頁面變更提醒

A page-change alert is useful only when an editor can open the exact source that justified it. Short answer: what a scrape plus embed pipeline looks like end to end is five separately retryable stages: discover pages, deduplicate canonical URLs, scrape readable text, chunk and embed it, then upsert vectors with the source URL in metadata.

For a media market-research tool, I would optimize for evidence before throughput. A fast alert with no retrievable URL is an assertion. A slower alert that identifies the changed page, the old and new content hashes, and the affected chunks can be checked and corrected. That distinction determines the queue boundaries, counters, and storage schema.

What does a scrape plus embed pipeline look like under failure?

Imagine a scheduled watch of 600 publisher and company pages. Search finds 87 candidates, URL normalization reduces them to 64, 61 fetch successfully, 58 produce readable text, and only 55 reach the vector index before a worker loses its lease. The numbers are illustrative, not a benchmark. Their purpose is operational: a single job succeeded metric would hide three extraction losses and three unfinished writes.

The invariant is stricter: every accepted source must end in one of two states, indexed with its URL or recorded as a named failure at a named stage. There is no miscellaneous gap.

No exceptions.

This is where a scrape-and-embed pipeline starts to look like delivery infrastructure. Search can be repeated. Fetches time out. Parsers can return empty copy. An embedding batch may be rejected. An upsert can succeed while the worker fails before acknowledging its queue message. Each boundary therefore needs its own attempt count, success count, failure count, and retry policy.

Retries stay local.

Do canonical URL deduplication before fetching. Tracking parameters, fragments, and equivalent links otherwise consume fetch capacity and may create duplicate chunks. Keep the original discovered URL for diagnosis, but use the canonical URL as the source identity. If a publisher declares a canonical URL in the document, reconcile it under an explicit policy; silently changing identity halfway through a run makes incident review painful.

The indexing key should also be deterministic. A practical key is a digest of canonical URL, normalized content hash, and chunk ordinal. Replaying an at-least-once queue delivery then overwrites the same records instead of multiplying them.

The five-stage contract

The stages need small, inspectable handoffs:

Extraction and chunking share a row because they usually share a worker, but they remain distinct counters. Empty readable text is not a successful zero-chunk document. Stop it there and classify it.

Metadata is part of the retrieval contract. At minimum, retain the canonical source URL, fetch time, document content hash, chunk ordinal, and a locator such as a heading or character range. The URL is non-negotiable: it lets the alerting layer cite the evidence and lets an operator trace a surprising retrieval result back to the page. A vector without provenance is difficult to audit and unsafe to present as support for a reported change.

This creates a deliberate trade-off. More provenance makes each record larger, but omitting it turns a later retrieval dispute into guesswork. For market research, I would accept the metadata overhead because an editor must be able to inspect the claim, and because stale or misattributed evidence costs more operational attention than a few additional fields. The decision changes for disposable similarity search where no result will be quoted or alerted on.

Chunk boundaries should be stable under small edits. Fixed character windows are easy to implement, but an inserted paragraph can shift every later window and manufacture a page-wide diff. Heading- or paragraph-aware chunks usually localize change better for articles and press pages. Their trade-off is parser dependence; malformed markup can produce uneven chunks. Record the chosen chunker version so a deliberate rechunk is distinguishable from a source edit.

Prevent duplicate alerts, not just duplicate vectors

Before wiring requests, inspect the contract you are about to call. This runnable Go program reads Infrai's public discovery document through the required Bearer convention, handles rate limiting, rejects non-2xx responses, and prints the declared method and path for each available web or vector capability. It uses one discovery route; it does not guess request bodies that belong in the returned JSON Schema.

package main

import (
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

type Capability struct {
    ID        string `json:"id"`
    Module    string `json:"module"`
    Method    string `json:"method"`
    Path      string `json:"path"`
    Available bool   `json:"available"`
}

type Discovery struct {
    Version      string       `json:"version"`
    GeneratedAt  string       `json:"generated_at"`
    Capabilities []Capability `json:"capabilities"`
}

func main() {
    key := os.Getenv("INFRAI_API_KEY")
    if key == "" {
        fmt.Fprintln(os.Stderr, "INFRAI_API_KEY is required")
        os.Exit(2)
    }
    baseURL := strings.TrimRight(os.Getenv("INFRAI_BASE_URL"), "/")
    if baseURL == "" {
        fmt.Fprintln(os.Stderr, "INFRAI_BASE_URL is required")
        os.Exit(2)
    }

    client := &http.Client{Timeout: 30 * time.Second}
    var response *http.Response
    for attempt := 0; attempt < 5; attempt++ {
        req, err := http.NewRequest(http.MethodGet, baseURL+"/discovery", nil)
        if err != nil {
            panic(err)
        }
        req.Header.Set("Authorization", "Bearer "+key)
        response, err = client.Do(req)
        if err != nil {
            panic(err)
        }
        if response.StatusCode != http.StatusTooManyRequests {
            break
        }
        response.Body.Close()
        delay := time.Duration(1<<attempt) * time.Second
        if seconds, err := strconv.Atoi(response.Header.Get("Retry-After")); err == nil {
            delay = time.Duration(seconds) * time.Second
        }
        time.Sleep(delay)
    }
    if response == nil {
        panic("discovery request produced no response")
    }
    defer response.Body.Close()
    if response.StatusCode < 200 || response.StatusCode >= 300 {
        body, _ := io.ReadAll(io.LimitReader(response.Body, 4096))
        panic(fmt.Sprintf("discovery: %s: %s", response.Status, body))
    }

    var document Discovery
    if err := json.NewDecoder(response.Body).Decode(&document); err != nil {
        panic(err)
    }
    for _, capability := range document.Capabilities {
        if capability.Available && (capability.Module == "web" || capability.Module == "vector") {
            fmt.Printf("%s\t%s\t%s\n", strings.ToUpper(capability.Method), capability.Path, capability.ID)
        }
    }
}

The worker path below captures the next important behavior without tying extraction or vector storage to one provider. It canonicalizes before fetching, derives stable record IDs, preserves the source URL, and makes the final alert conditional on a content-hash transition. Implementations behind the interfaces apply retries to their own network calls and expose per-stage counters.

package pipeline

import (
    "context"
    "crypto/sha256"
    "encoding/hex"
    "fmt"
    "net/url"
    "sort"
    "strings"
    "time"
)

type Page struct {
    CanonicalURL string
    Text         string
}

type Record struct {
    ID       string
    Vector   []float32
    Metadata map[string]string
}

type Fetcher interface {
    Fetch(context.Context, string) (Page, error)
}

type Embedder interface {
    Embed(context.Context, []string) ([][]float32, error)
}

type Index interface {
    Upsert(context.Context, []Record) error
}

type State interface {
    SwapHash(context.Context, string, string) (previous string, err error)
}

func canonical(raw string) (string, error) {
    u, err := url.Parse(raw)
    if err != nil || u.Host == "" {
        return "", fmt.Errorf("parse URL %q: %w", raw, err)
    }
    u.Fragment = ""
    q := u.Query()
    for key := range q {
        if strings.HasPrefix(strings.ToLower(key), "utm_") {
            q.Del(key)
        }
    }
    u.RawQuery = q.Encode()
    u.Host = strings.ToLower(u.Host)
    return u.String(), nil
}

func digest(parts ...string) string {
    h := sha256.Sum256([]byte(strings.Join(parts, "\x00")))
    return hex.EncodeToString(h[:])
}

func Process(ctx context.Context, rawURLs []string, fetcher Fetcher, embedder Embedder, index Index, state State) ([]string, error) {
    unique := map[string]struct{}{}
    for _, raw := range rawURLs {
        u, err := canonical(raw)
        if err != nil {
            return nil, err
        }
        unique[u] = struct{}{}
    }
    urls := make([]string, 0, len(unique))
    for u := range unique {
        urls = append(urls, u)
    }
    sort.Strings(urls)

    changed := make([]string, 0)
    for _, u := range urls {
        page, err := fetcher.Fetch(ctx, u)
        if err != nil {
            return changed, fmt.Errorf("fetch %s: %w", u, err)
        }
        chunks := chunkParagraphs(page.Text, 1200)
        if len(chunks) == 0 {
            return changed, fmt.Errorf("extract %s: no readable text", u)
        }
        docHash := digest(page.Text)
        vectors, err := embedder.Embed(ctx, chunks)
        if err != nil {
            return changed, fmt.Errorf("embed %s: %w", u, err)
        }
        if len(vectors) != len(chunks) {
            return changed, fmt.Errorf("embed %s: got %d vectors for %d chunks", u, len(vectors), len(chunks))
        }

        records := make([]Record, len(chunks))
        for i := range chunks {
            records[i] = Record{
                ID:     digest(u, docHash, fmt.Sprint(i)),
                Vector: vectors[i],
                Metadata: map[string]string{
                    "source_url":   u,
                    "content_hash": docHash,
                    "chunk_ordinal": fmt.Sprint(i),
                    "fetched_at":   time.Now().UTC().Format(time.RFC3339),
                },
            }
        }
        if err := index.Upsert(ctx, records); err != nil {
            return changed, fmt.Errorf("upsert %s: %w", u, err)
        }
        previous, err := state.SwapHash(ctx, u, docHash)
        if err != nil {
            return changed, fmt.Errorf("commit state %s: %w", u, err)
        }
        if previous != "" && previous != docHash {
            changed = append(changed, u)
        }
    }
    return changed, nil
}

func chunkParagraphs(text string, limit int) []string {
    paragraphs := strings.Split(strings.TrimSpace(text), "\n\n")
    chunks := make([]string, 0, len(paragraphs))
    for _, paragraph := range paragraphs {
        paragraph = strings.TrimSpace(paragraph)
        for len(paragraph) > limit {
            chunks = append(chunks, paragraph[:limit])
            paragraph = strings.TrimSpace(paragraph[limit:])
        }
        if paragraph != "" {
            chunks = append(chunks, paragraph)
        }
    }
    return chunks
}

One ordering detail matters: commit the new page hash only after the index acknowledges the records. Otherwise an index failure can advance state, and the next run will conclude that nothing changed. Alert delivery needs its own idempotency key, such as the canonical URL plus old and new hashes, because a retry after an uncertain acknowledgment must not notify an editor twice.

State follows data.

The sample stops on the first failed page so its failure semantics are obvious. A production fan-out would isolate pages, bound concurrency, and retry each stage independently. On HTTP 429, it should honor Retry-After when supplied and otherwise use capped exponential backoff with jitter. Persistent failures belong in a reviewable dead-letter path with the stage, URL, attempt count, and last error attached.

Comparing the assembly choices fairly

There is no universal winner because these products own different parts of the path.

Firecrawl is oriented around turning web content into LLM-ready data and documents scraping, crawling, mapping, and search. It is a focused choice when web acquisition and extraction quality are the difficult part. You still need to define index identity, vector metadata, change state, and alert idempotency around it.

Apify provides Actors and storage for running web automation and scraping workloads. It fits teams that need a broad automation runtime or an existing Store actor for a particular site. The operational boundary is wider than a scrape request, so evaluate actor behavior, dataset retention, scheduling, and downstream indexing as separate contracts.

Unstructured focuses on partitioning and chunking many document types. It is compelling when PDFs, office files, email, and HTML must enter one normalization path. It does not remove the need for URL discovery or an idempotent vector write, and its richer element model asks you to make an explicit mapping into retrieval metadata.

Pinecone is a managed vector database with upsert and metadata filtering. It is a natural destination when retrieval operations and index management deserve a dedicated service. It does not fetch or extract pages, which can be an advantage: acquisition failures remain visibly outside the index rather than being blurred into one platform job.

Weaviate and Qdrant also provide vector search with metadata filtering, while pgvector keeps vectors beside relational data in PostgreSQL. Weaviate fits a team that wants a dedicated vector database with an object-oriented data model; Qdrant is attractive when payload filtering and a dedicated vector service match the retrieval design. pgvector reduces the number of state systems for a team already operating PostgreSQL, though database capacity and index tuning remain that team's responsibility. These three are indexing choices, not scraping products, so pair them with an acquisition and extraction layer.

Infrai is another valid assembly choice when a small team values one REST API and one key across search, scraping, and vector operations, without installing a separate SDK for each capability. Its public discovery endpoint is self-describing: capability details include request and response schemas, billing information, and runnable examples, so wiring a new capability starts with inspecting the endpoint. Keep the same stage-level metrics anyway; a unified API does not make five failure domains into one.

My decision rule is blunt. Choose Firecrawl when web extraction is the center of risk, Apify when custom web automation is the job, Unstructured when heterogeneous document parsing dominates, and Pinecone when vector operations need an independently managed boundary. Consider the unified surface when integration sprawl is the bigger constraint. A mixed stack is reasonable, provided one component owns each durable handoff and the source URL survives all of them.

When this design is too much

Do not build a vector pipeline for a dozen fixed pages if exact DOM or text diffs answer the question. Conditional GETs, content hashes, and a small relational table may yield clearer alerts with fewer moving parts. Embeddings earn their place when editors need semantic retrieval across a growing corpus, when wording changes but meaning remains relevant, or when the alert must include related prior evidence.

Also avoid treating every hash change as editorial news. Cookie banners, timestamps, rotating recommendations, and navigation can change while the article body does not. Extraction scope is therefore part of correctness. Store enough provenance to explain what was retained, but do not claim the parser found the meaningful change until a human or an evaluation set verifies that policy.

The runbook test is simple: given an alert ID, can an operator identify the canonical page, both content hashes, the changed chunks, every failed or retried stage, and the exact vector records written? If any link is missing, the pipeline may retrieve text, but it cannot yet support a citable media alert.

Sources

• Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks
• Firecrawl documentation
• Apify Actors documentation
• Unstructured partitioning documentation
• Pinecone upsert documentation
• Weaviate documentation
• Qdrant documentation
• pgvector documentation
• RFC 9110: Retry-After