用网页数据构建知识图谱用 LLM 把公司网站抓取进 Neo4j

git clone https://github.com/ScraperHub/build-a-knowledge-graph-from-crawled-websites-with-neo4j.git
cd build-a-knowledge-graph-from-crawled-websites-with-neo4j
python -m venv .venv
source .venv/bin/activate
cp steps/.env.example steps/.env
cp final/.env.example final/.env

CRAWLBASE_TOKEN=YOUR_CRAWLBASE_TOKEN
CRAWLBASE_JS_TOKEN=YOUR_CRAWLBASE_JS_TOKEN
OPENAI_API_KEY=YOUR_OPENAI_API_KEY
OPENAI_MODEL=gpt-4o-mini
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=YOUR_NEO4J_PASSWORD

pip install -r steps/requirements.txt
cd final && docker compose up -d --wait

PAGE_TYPES = {
    "about": r"^(about|company|our-story)",
    "team": r"^(team|our-team|leadership|people)",
    "partners": r"^partner",
    "integrations": r"^(integration|marketplace)",
    "careers": r"^(career|jobs)",
}
MAX_PAGES = 8


def host(url: str) -> str:
    return (urlsplit(url).hostname or "").lower().removeprefix("www.")


def page_type(url: str) -> str | None:
    segments = [s for s in urlsplit(url).path.lower().split("/") if s]
    for name, pattern in PAGE_TYPES.items():
        if segments and re.match(pattern, segments[-1]):
            return name
    return None

Basecamp: 52 links, 2 company pages (landed on https://basecamp.com/)
  about         https://basecamp.com/about
  integrations  https://basecamp.com/integrations
HEY: 54 links, 0 company pages (landed on https://www.hey.com/)
Pipedrive: 103 links, 4 company pages (landed on https://www.pipedrive.com/)
  integrations  https://www.pipedrive.com/en/marketplace
  about         https://www.pipedrive.com/en/about
  careers       https://www.pipedrive.com/en/jobs
  partners      https://www.pipedrive.com/en/partner
Zendesk: 125 links, 4 company pages (landed on https://www.zendesk.com/)
  integrations  https://www.zendesk.com/marketplace
  ...

def fetch_page(url: str) -> dict:
    """Readability first, plain Markdown if that drops the content, JS token if the page renders in the browser."""
    for token, readability, min_chars in ((CRAWLBASE_TOKEN, True, 1500),
                                          (CRAWLBASE_TOKEN, False, 300),
                                          (CRAWLBASE_JS_TOKEN, False, 0)):
        response = fetch(url, token, readability)
        if len(response.text) >= min_chars:
            break
    rid = httpx.URL(response.headers.get("storage_url", "")).params.get("rid", "")
    return {"url": url, "markdown": response.text, "rid": rid, "readability": readability,
            "js": token == CRAWLBASE_JS_TOKEN}

Basecamp     3253 chars  readability  rid=abac8926164450fab0456a27  https://basecamp.com/about
Basecamp    16676 chars  plain        rid=86c17c449811ab0e0f861dd3  https://basecamp.com/integrations
Pipedrive   17742 chars  plain        rid=d2ef101d7ac8d30bf3a26a5a  https://www.pipedrive.com/en/marketplace
Pipedrive   11187 chars  plain        rid=d8d8e67054e10a90c81fa2cb  https://www.pipedrive.com/en/about
Zendesk      8267 chars  plain        rid=78b15c345cde1256878b45bd  https://www.zendesk.com/company

class Evidence(BaseModel):
    model_config = ConfigDict(extra="forbid")
    name: str = Field(min_length=1)
    source_quote: str = Field(min_length=3, description="Exact words copied from the page.")


class Company(Evidence): pass
class Product(Evidence): pass
class Integration(Evidence): pass


class Person(Evidence):
    title: str | None
    employer: str | None


class Location(Evidence):
    country: str | None


class Relationship(BaseModel):
    model_config = ConfigDict(extra="forbid")
    source: str
    type: Literal["WORKS_AT", "OFFERS", "INTEGRATES_WITH", "PARTNERS_WITH", "LOCATED_IN"]
    target: str
    source_quote: str = Field(min_length=3)

Good output parses: Paulo Cunha
Bad output rejected with 2 errors:
   people.0.source_quote -> Field required
   relationships.0.type -> Input should be 'WORKS_AT', 'OFFERS', 'INTEGRATES_WITH', 'PARTNERS_WITH' or 'LOCATED_IN'

def guard(result, page: str, company: str):
    """Keep entities whose quote is on the page and relationships whose endpoints survived."""
    page_text = normalize(page)
    found = lambda quote: len(quote) > 2 and normalize(quote).rstrip(".,;:!?") in page_text  # noqa: E731
    dropped = []
    for field in ("companies", "people", "products", "integrations", "locations"):
        items = getattr(result, field)
        dropped += [f"{field}: {e.name}" for e in items if not found(e.source_quote)]
        setattr(result, field, [e for e in items if found(e.source_quote)])

https://basecamp.com/integrations: 19 relationships, 4 dropped
   dropped integrations: Trello
   dropped locations: Chicago
   dropped relationship: Basecamp INTEGRATES_WITH Trello
   dropped relationship: Basecamp LOCATED_IN Chicago
https://www.pipedrive.com/en/partner: 1 relationships, 1 dropped
   dropped relationship: Pipedrive PARTNERS_WITH Zimple

def key(name: str) -> str:
    """"Stripe, Inc." and "stripe.com" both become "stripe"."""
    name = re.sub(r"\.(com|io|ai|co|app)$", "", name.lower().strip())
    return " ".join(t for t in re.sub(r"[^\w\s&+]", " ", name).split() if t not in SUFFIXES)

'Stripe' vs 'Stripe, Inc.': score 100, merged=True
'Salesforce' vs 'Salesforce.com': score 100, merged=True
'Notion' vs 'Motion': score 83, merged=False

NODE = """
UNWIND $rows AS row
MERGE (n:{label} {{name: row.name}})
SET n += row.props,
    n.source_urls = CASE WHEN row.url IN coalesce(n.source_urls, []) THEN n.source_urls
                    ELSE coalesce(n.source_urls, []) + row.url END,
    n.rids = CASE WHEN row.rid IN coalesce(n.rids, []) THEN n.rids ELSE coalesce(n.rids, []) + row.rid END
"""

REL = """
UNWIND $rows AS row
MATCH (a:{source} {{name: row.source}}), (b:{target} {{name: row.target}})
MERGE (a)-[r:{type} {{source_url: row.url}}]->(b)
SET r.rid = row.rid, r.extracted_at = row.extracted_at, r.source_quote = row.source_quote
"""

After load 1: 90 nodes, 119 relationships
After load 2: 90 nodes, 119 relationships

QUERIES = {
    "Shared integrations": """
        MATCH (a:Company)-[:INTEGRATES_WITH]->(i:Integration)<-[:INTEGRATES_WITH]-(b:Company)
        WHERE a.name < b.name
        RETURN a.name AS company_a, b.name AS company_b,
               count(DISTINCT i) AS shared, collect(DISTINCT i.name)[..6] AS integrations
        ORDER BY shared DESC LIMIT 5""",
    "People on two companies' sites": """
        MATCH (p:Person)-[r:WORKS_AT]->(c:Company)
        WITH p, collect(DISTINCT c.name) AS companies, collect(DISTINCT r.source_url) AS pages
        WHERE size(companies) >= 2
        RETURN p.name AS person, companies, pages""",
    "Geographic clusters": """
        MATCH (c:Company)-[:LOCATED_IN]->(l:Location)
        WITH l.country AS country, collect(DISTINCT c.name) AS companies, collect(DISTINCT l.name) AS cities
        WHERE size(companies) >= 2
        RETURN country, companies, cities ORDER BY size(companies) DESC""",
}

Shared integrations:
   {"company_a": "Basecamp", "company_b": "Pipedrive", "shared": 1, "integrations": ["Zapier"]}

People on two companies' sites:
   {"person": "Jason Fried", "companies": ["HEY", "Basecamp"], "pages": ["https://www.hey.com/", "https://basecamp.com/", "https://basecamp.com/about"]}

Geographic clusters:
   {"country": "Germany", "companies": ["Pipedrive", "Zendesk"], "cities": ["Berlin"]}
   {"country": "Portugal", "companies": ["Pipedrive", "Zendesk"], "cities": ["Lisbon"]}

WRITE_WORDS = re.compile(r"\b(CREATE|MERGE|DELETE|DETACH|SET|REMOVE|DROP|LOAD|FOREACH|CALL)\b", re.I)


class Query(BaseModel):
    cypher: str


def check_read_only(cypher: str) -> str:
    """Reject write clauses. String literals are blanked first so names like 'Merge Labs' pass."""
    code = re.sub(r"'[^']*'|\"[^\"]*\"", "''", cypher).strip().rstrip(";")
    if ";" in code or WRITE_WORDS.search(code) or not re.match(r"(?i)\s*(MATCH|OPTIONAL|WITH|UNWIND)", code):
        raise ValueError(f"Blocked query: {cypher}")
    return cypher.strip().rstrip(";")

with GraphDatabase.driver(NEO4J_URI, auth=NEO4J_AUTH) as driver, driver.session() as session:
    rows = session.execute_read(lambda tx: [r.data() for r in tx.run(cypher)])

Safety check works: Blocked query: MATCH (n) DETACH DELETE n

Cypher:
MATCH (a:Company)-[r1:INTEGRATES_WITH]->(i)<-[r2:INTEGRATES_WITH]-(b:Company)
WHERE toLower(a.name) = toLower('Basecamp') AND toLower(b.name) = toLower('Pipedrive')
UNWIND [r1, r2] AS r
RETURN i.name AS integration, startNode(r).name AS company, r.source_url AS source_url, r.rid AS rid LIMIT 25

Rows: 2

Answer:
Basecamp and Pipedrive have the integration "Zapier" in common. This is supported by the following sources: [https://basecamp.com/integrations, rid: 86c17c449811ab0e0f861dd3] and [https://www.pipedrive.com/en/marketplace, rid: d2ef101d7ac8d30bf3a26a5a].

def get_stored(rid: str) -> str:
    response = httpx.get(STORAGE, params={"token": CRAWLBASE_TOKEN, "rid": rid}, timeout=60)
    response.raise_for_status()
    print(f"  rid={rid} stored_at={response.headers.get('stored_at')} url={response.headers.get('url')}")
    return response.text

rid=abac8926164450fab0456a27 stored_at=2026-10-29T10:16:17+01:00 url=https://basecamp.com/about
re-extracted 2 relationships, dropped 0
rid=86c17c449811ab0e0f861dd3 stored_at=2026-10-29T10:16:33+01:00 url=https://basecamp.com/integrations
re-extracted 20 relationships, dropped 2

python -m pip install -r requirements.txt
python -m pip install -e .
python -m kg_pipeline run-all --seeds data/seed_small.csv

Fetched 14 pages. Saved to .../final/output/pages.json
Extracted 14 pages: kept 223 items, dropped 11 by the quote guard.
Merged 0 near-duplicate names.
Graph now has 130 nodes and 159 relationships.

python -m kg_pipeline ask "Which integrations do Basecamp and Pipedrive have in common?"
python -m kg_pipeline refresh
python -m kg_pipeline load --seeds data/seed_small.csv

python -m pytest -m "not integration"