This document specifies the browser-use harvesting layer that feeds the institutional investor graph database documented in sqlite-graph-db-investor-relationships-2026.md. It covers library architecture, task specification patterns, target site playbooks, integration with the Chandra OCR pipeline, and the operational scheduler design.
The core thesis: browser-use turns LLM-directed browsing into a structured data extraction pipeline. Rather than writing brittle CSS selectors that break on site redesigns, browser-use tasks describe intent — “extract the names and titles of all board members” — and the underlying Playwright-backed agent handles DOM traversal, pagination, and dynamic content loading. When the site changes structure, the task description still works; a CSS scraper does not.
The output of every harvest run is a list of JSON records that map directly to the nodes and edges tables in the SQLite graph schema. The Chandra OCR pipeline is triggered as a side effect when the agent discovers PDF links (annual reports, board meeting minutes) during traversal.
Part 1 — Browser-Use Library Architecture
1.1 Core Components
browser-use is a Python library (github.com/browser-use/browser-use) that wraps Playwright with an LLM-directed agent layer. The primary abstractions:
Agent — the top-level orchestrator. Takes a task string, an LLM instance, and a Browser instance. Runs an action loop: observe page state → decide next action → execute via Controller → check if done. The agent exposes await agent.run() which returns an AgentHistoryList containing all actions taken and the final extracted result.
Controller — the action registry. Defines what actions are available: go_to_url, click_element, input_text, extract_content, scroll_down, wait_for_element, save_pdf, done. Controllers can be extended with custom actions — this is how you inject a “save to graph DB” action that fires mid-task rather than waiting for final output.
Browser / BrowserContext — thin wrappers around Playwright’s browser and context objects. The Browser class handles browser lifecycle; BrowserContext manages cookies, authentication state, and page isolation. Authentication sessions can be persisted as browser state files, enabling authenticated harvests without re-logging in on each run.
BrowserConfig — configuration dataclass for the browser instance. Key fields: headless (bool), user_agent (str), extra_chromium_args (list), proxy (ProxySettings), slow_mo (int, milliseconds between actions for anti-detection).
Task specification — a plain-English string that describes what to extract, where to find it, and what format to return it in. The LLM interprets this string; no DOM selectors required. The quality of the output scales directly with task specificity.
1.2 LLM Backend Selection
browser-use supports any LangChain-compatible LLM. For this pipeline:
- Claude Sonnet 4.5 / 4.6 — best reasoning for ambiguous DOM structures; recommended for Tier 1 targets where extraction logic requires judgment (e.g., distinguishing trustee from staff in a mixed-role page). Cost: ~$3–6 per 1M input tokens, ~$15 per 1M output tokens.
- GPT-4o — comparable quality; slightly faster on simple extraction tasks. Use as fallback.
- Gemini 2.0 Flash — lowest cost for high-volume, simple extraction tasks (staff directories with consistent table structure). Use for Tier 2 targets.
For the harvesting pipeline, configure Claude Sonnet 4.6 as primary and Gemini 2.0 Flash as cost fallback for sites with predictable structure:
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI
def get_llm(tier: str = "primary") -> object:
"""Return LLM appropriate for harvest tier."""
if tier == "primary":
return ChatAnthropic(
model="claude-sonnet-4-6",
temperature=0,
max_tokens=4096
)
else:
return ChatGoogleGenerativeAI(
model="gemini-2.0-flash",
temperature=0,
max_output_tokens=2048
)
1.3 Handling Authentication
Three authentication patterns appear across target sites:
Pattern A — No auth required (public pages). Most public pension rosters, endowment IC pages, and conference speaker pages are unauthenticated. Pass no auth configuration; browser-use navigates directly.
Pattern B — Form-based login with persistent session. Sites like pension member portals that gate full rosters behind login. Approach: run an initial setup agent that logs in and saves browser state; subsequent harvest agents load that state instead of re-authenticating.
from browser_use import Agent, Browser, BrowserConfig, BrowserContext
async def login_and_save_state(
login_url: str,
username: str,
password: str,
state_path: str
):
"""Run once to capture authenticated browser state."""
browser = Browser(config=BrowserConfig(headless=False)) # headless=False for CAPTCHA-prone sites
context = await browser.new_context()
page = await context.new_page()
await page.goto(login_url)
await page.fill('[name=username]', username)
await page.fill('[name=password]', password)
await page.click('[type=submit]')
await page.wait_for_load_state('networkidle')
await context.storage_state(path=state_path)
await browser.close()
async def authenticated_harvest(
task: str,
state_path: str,
llm
) -> dict:
"""Harvest using saved auth state."""
browser = Browser(config=BrowserConfig(headless=True))
context = await browser.new_context(storage_state=state_path)
agent = Agent(task=task, llm=llm, browser_context=context)
result = await agent.run()
await browser.close()
return result.final_result()
Pattern C — Conference/event registration portals (iConnections, NCPERS, NACUBO). These often use SSO or email-link login. Use headless=False for the initial login run; save state after manual CAPTCHA completion. Most conference speaker pages are publicly accessible without login; the member directory behind login is the secondary target.
1.4 Handling Dynamic/JS-Heavy Sites
Salesforce-based portals and React/Vue SPAs load content asynchronously. browser-use handles this automatically because Playwright waits for networkidle by default before each action. Supplement with explicit waits in the task description:
Task: "Navigate to https://example-pension.org/board. Wait for the page to fully load including any dynamically rendered sections. Then extract all names and titles of current board members..."
For Salesforce-based consultant portals that use Visualforce or Lightning components:
from browser_use import Controller
# Custom action: wait for Lightning component render
controller = Controller()
@controller.action("Wait for Salesforce Lightning render")
async def wait_for_lightning(browser_context):
"""Waits for Salesforce Lightning components to finish rendering."""
page = browser_context.page
await page.wait_for_selector('.slds-card, .forceOutputLookup, [data-component-id]',
timeout=15000)
await page.wait_for_load_state('networkidle')
return "Lightning render complete"
1.5 Anti-Detection Considerations
Public pension pages and conference speaker directories are public records — aggressive anti-bot measures are unusual. Apply baseline courtesy regardless:
Rate limiting: Insert slow_mo=1500 (1.5 seconds between actions) in BrowserConfig. Add explicit sleep between page loads: await asyncio.sleep(random.uniform(2, 5)). Never hammer a site faster than a human would browse.
User agent rotation: Maintain a pool of realistic user agents and rotate per harvest run:
import random
USER_AGENTS = [
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Safari/605.1.15",
]
def get_browser_config(proxy: str = None) -> BrowserConfig:
config = BrowserConfig(
headless=True,
user_agent=random.choice(USER_AGENTS),
slow_mo=random.randint(1000, 2000),
extra_chromium_args=["--disable-blink-features=AutomationControlled"]
)
if proxy:
config.proxy = {"server": proxy}
return config
Residential proxy integration: For Tier 2 targets (consultant staff directories, LinkedIn Google-dork results) that may have bot detection, route through a residential proxy pool. Oxylabs, Bright Data, and Smartproxy offer Python SDK integration:
# Proxy rotation per-session via Bright Data
PROXY_CONFIG = {
"server": "https://zproxy.lum-superproxy.io:22225",
"username": "lum-customer-YOUR_ID-zone-residential",
"password": "YOUR_PASSWORD"
}
config = BrowserConfig(
headless=True,
proxy=PROXY_CONFIG,
slow_mo=2000
)
Robots.txt compliance: Before adding any site to the harvest queue, check robots.txt. If the target path is disallowed, remove it from automated harvest and flag for manual collection.
import urllib.robotparser
def check_robots(base_url: str, target_path: str, user_agent: str = "*") -> bool:
"""Returns True if crawling is permitted."""
rp = urllib.robotparser.RobotFileParser()
rp.set_url(f"{base_url}/robots.txt")
rp.read()
return rp.can_fetch(user_agent, f"{base_url}{target_path}")
1.6 Output Format: JSON → SQLite Graph Schema
Every browser-use task must return a JSON object with the following standard envelope. The harvester function validates against this schema before writing to the graph DB:
from dataclasses import dataclass, field
from typing import Optional, List
@dataclass
class HarvestPerson:
"""A person extracted from a web page — maps to a `nodes` row."""
name: str
title: Optional[str] = None
organization: Optional[str] = None
email: Optional[str] = None
phone: Optional[str] = None
bio_url: Optional[str] = None
source_url: str = ""
# Graph schema fields
node_type: str = "person"
node_subtype: Optional[str] = None # trustee|committee_member|staff|consultant
confidence: float = 0.70 # web_scrape default
@dataclass
class HarvestOrganization:
"""An organization extracted from a web page — maps to a `nodes` row."""
name: str
canonical_id: Optional[str] = None # EIN:xxx or CRD:xxx if determinable
url: str = ""
source_url: str = ""
node_type: str = "firm"
node_subtype: Optional[str] = None # public_pension|endowment|foundation|advisory_firm
confidence: float = 0.75
@dataclass
class HarvestRelationship:
"""A relationship extracted from a web page — maps to an `edges` row."""
person_name: str
organization_name: str
edge_type: str # TRUSTEE_OF|COMMITTEE_MEMBER_OF|WORKS_AT
title: Optional[str] = None
valid_from: Optional[str] = None
confidence: float = 0.70
@dataclass
class HarvestResult:
"""Standard output envelope from every browser-use harvest task."""
source_url: str
harvest_timestamp: str
organization: Optional[HarvestOrganization] = None
persons: List[HarvestPerson] = field(default_factory=list)
relationships: List[HarvestRelationship] = field(default_factory=list)
pdf_links: List[str] = field(default_factory=list) # PDFs to queue for Chandra
errors: List[str] = field(default_factory=list)
raw_llm_output: str = ""
The task prompt must request this JSON structure explicitly. See Part 4 for full task examples.
Part 2 — Priority Target Sites
Tier 1 — High-Value, Permissive
2.1 Public Pension Board Rosters
All major US public pensions publish board member rosters under state sunshine laws. These are the highest-quality free harvest targets: no auth required, publically mandated disclosure, trustee names are exactly what we need.
| Fund | AUM | Roster URL | Update frequency |
|---|---|---|---|
| CalPERS | $502B | calpers.ca.gov/about/organization/board-of-administration | Continuous |
| CalSTRS | $340B | calstrs.com/board-of-directors | Continuous |
| NYCDRS / NYCERS | $80B | nycers.org/about-nycers/board-of-trustees | Continuous |
| TRSIL | $68B | trs.illinois.gov/about/board | Continuous |
| WSIB | $195B | sib.wa.gov/board | Continuous |
| OTRS (Oklahoma TRS) | $19B | ok.gov/trs/About_Us/Board_of_Trustees | Periodic |
| Texas TRS | $205B | trs.texas.gov/trs/about/trustees.html | Continuous |
| LACERA | $88B | lacera.com/board | Continuous |
| Ohio STRS | $90B | strsoh.org/employer/board | Continuous |
| Florida SBA | $220B | myfloridacffo.com | Periodic |
Annual report PDFs for each of these funds contain investment staff names (CIO, deputy CIOs, portfolio managers by asset class). The board roster page often links to these PDFs — queue them for Chandra.
Harvest cadence: Monthly. Board composition changes slowly (terms are multi-year); monthly is sufficient to catch appointments and resignations within a reasonable window.
2.2 University Endowment Investment Committee Pages
Harvard, Yale, MIT, Stanford, Princeton, Duke, Penn, Columbia, Northwestern, and Michigan endowments have dedicated investment offices with public-facing pages. Some publish IC member names; others publish only the CIO. The IC member disclosure level varies:
- High disclosure (IC names published): MIT Investment Management Company (mitimco.org), Princeton University Investment Company (princetoninvestment.com), Duke Management Company (dumac.org), Penn’s UPENN Investment Office
- CIO-only disclosure: Harvard Management Company (hmc.harvard.edu), Yale Investments Office (investments.yale.edu), Stanford Management Company (smc.stanford.edu)
For CIO-only disclosures, supplement with Form 990 Part VII which names compensated officers including the CIO, and with conference speaker bios (P&I, NACUBO) which routinely include IC affiliation.
Harvest cadence: Quarterly for IC pages; semi-annually for 990-based CIO extraction.
2.3 Foundation Investment Committee Pages
Large private and community foundations often publish governance pages naming trustees and IC members.
| Foundation | AUM | Disclosure level |
|---|---|---|
| Bill & Melinda Gates Foundation | $67B | CIO named; IC structure disclosed |
| Ford Foundation | $16B | Investment committee named |
| Rockefeller Foundation | $6B | Trustees list published |
| MacArthur Foundation | $8B | Board and investment committee disclosed |
| Hewlett Foundation | $15B | Board roster published |
| Packard Foundation | $9B | Board roster published |
| Wellcome Trust (UK) | £38B | Investment committee published |
| Robert Wood Johnson | $14B | Trustees disclosed |
| Kellogg Foundation | $8B | Trustees disclosed |
Data point: Foundation investment committee members are frequently the same people who serve on corporate boards, university boards, and other foundations — connecting them in the graph creates a high-value relationship map for identifying shared influence nodes.
Harvest cadence: Quarterly.
2.4 SEC EDGAR Full-Text Search — Form ADV
The EDGAR full-text search API (efts.sec.gov) is the richest free source for investment professional names across thousands of advisory firms. Two primary extraction targets:
Form ADV Schedule D Section 7.A — control persons and executive officers. Every SEC-registered RIA must file these; the data is machine-readable via the EDGAR structured data API.
Form ADV Part 2B Brochure Supplements — individual supervised persons with 5-year employment history. These are PDF filings with consistent structure across all advisers; Chandra OCR handles them well after download.
browser-use task: navigate to EDGAR full-text search, query for a target firm, locate the latest ADV-W or ADV filing, download the brochure supplements (Part 2B), queue for Chandra, and extract structured officer data from the Part 1A Schedule D HTML view.
Harvest cadence: Monthly for target consultant firms; quarterly broad sweep for new ADV filers in institutional investment space.
2.5 Conference Speaker Pages
P&I, NCPERS, iConnections, and NACUBO conference speaker bios are gold mines: they are self-reported, current, and always include organizational affiliation with specific title. A speaker at the P&I Investment Consulting conference who lists “Investment Committee Member, XYZ Foundation” is a confirmed, self-disclosed IC member.
Target conferences:
| Conference | Organizer | Typical attendee profile | Speaker page accessibility |
|---|---|---|---|
| P&I Investment Consulting | Pensions & Investments | Pension CIOs, consultants | Public |
| P&I Alternative Investments | Pensions & Investments | Pension alts staff, GPs | Public |
| iConnections Global Alts | iConnections | LP decision-makers (verified) | Member-only agenda; speakers often public |
| NCPERS Annual | NCPERS | Public pension trustees, exec directors | Public |
| NACUBO Annual | NACUBO | University endowment CIOs, CFOs | Public |
| Institutional Investor Forum | II | Pension/endowment CIOs | Public (top speakers) |
| SuperReturn | BC Events | PE/VC focused LPs and GPs | Public speakers list |
| CAASA Annual (Canada) | CAASA | Canadian pension/endowment staff | Public |
| IPEM (Europe) | IPEM | European LP/GP | Public |
Harvest cadence: 2 weeks before each conference (speaker bios are freshest) + 4 weeks after (agendas often updated with attendance confirmation).
Tier 2 — Requires More Care
2.6 Investment Consultant Staff Directories
NEPC, Callan, RVK, Verus, Meketa, Cambridge Associates, and Wilshire publish staff directories on their websites. These are legitimate public pages. They require more care because:
- These firms have more sophisticated IT infrastructure and may block scrapers
- The pages have varying structures — some use Salesforce Communities, some static HTML, some JavaScript-rendered React apps
- The relationship between staff and specific client engagements is not disclosed; only firm + title is harvested
Approach: Use browser-use with slow_mo=3000 and residential proxy. Harvest the team/people directory page, extract names and titles. Cross-reference extracted titles against the firm’s Form ADV Part 2B to verify and enrich employment histories.
Target pages:
- NEPC: nepc.com/team
- Callan: callan.com/people (requires JS render)
- RVK: rvkinc.com/team
- Verus: verusinvestments.com/team
- Meketa: meketa.com/team
- Cambridge Associates: cambridgeassociates.com/who-we-are/our-people (Salesforce-based, use custom Lightning wait)
Harvest cadence: Quarterly.
2.7 Family Office Association Member Directories
FOX (Family Office Exchange), Campden Wealth, and similar organizations maintain member directories. These are typically gated behind paid membership.
Approach: For publicly accessible portions (conference speaker bios, published research with author affiliations), harvest directly. For gated member directories, use the conference speaker page approach — family office principals who speak at FOX or Campden Wealth events are identified by name with organizational affiliation in the public-facing conference materials.
Harvest cadence: Semi-annually for public-facing materials; conference approach quarterly.
2.8 LinkedIn via Google Dork → Public Profile Pages
Direct LinkedIn scraping violates LinkedIn’s Terms of Service and triggers aggressive bot detection. The compliant approach:
- Use Google dork syntax to find public LinkedIn profile pages for known targets:
site:linkedin.com/in/ "investment committee" "CalPERS"orsite:linkedin.com/in/ "Chief Investment Officer" "endowment" - Google returns public LinkedIn profile URLs in search results
- browser-use navigates to those public profile pages (not the logged-in feed)
- Extract publicly disclosed employment history from the public profile view
This approach harvests only what a non-logged-in user can see — legally and technically the same as a human reading the page. It does not use LinkedIn’s API and does not require authentication.
LINKEDIN_DORK_TEMPLATE = (
'site:linkedin.com/in/ "{title}" "{organization}"'
)
Important constraint: Google rate-limits automated searches. Implement a 30–60 second delay between Google search requests and a daily cap of 100 Google queries per IP.
Harvest cadence: As-needed enrichment for specific targets; not a bulk harvest source.
Part 3 — Integration with Existing Pipelines
3.1 Browser-Use → Chandra PDF Pipeline
When the harvest agent encounters PDF links (annual reports, board meeting minutes, Form ADV Part 2B brochures), it adds them to a download queue rather than attempting to extract content directly. The Chandra OCR pipeline then processes them.
Integration pattern:
import asyncio
import aiohttp
from pathlib import Path
from datetime import datetime
PDF_DOWNLOAD_DIR = Path("research/19-agent-frameworks/raw/browser-harvest")
async def download_pdf(url: str, dest_dir: Path) -> Path:
"""Download a PDF from URL and return local path."""
dest_dir.mkdir(parents=True, exist_ok=True)
# Sanitize filename from URL
filename = url.split("/")[-1].split("?")[0]
if not filename.endswith(".pdf"):
filename = f"harvest_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
dest = dest_dir / filename
if dest.exists():
return dest # Already downloaded; skip
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=120)) as resp:
if resp.status == 200:
content = await resp.read()
dest.write_bytes(content)
return dest
def queue_for_chandra(pdf_paths: list[Path], queue_file: Path):
"""
Append PDF paths to the Chandra ingest queue file.
The main ingest_pdfs.py script processes this queue.
Queue file format: one absolute path per line, with a tab-separated
source URL and timestamp for provenance tracking.
"""
with queue_file.open("a") as f:
for p in pdf_paths:
ts = datetime.utcnow().isoformat() + "Z"
f.write(f"{p.resolve()}\t{ts}\n")
async def harvest_with_pdf_discovery(
task: str,
source_url: str,
llm,
db,
resolver,
chandra_queue: Path
):
"""
Run a harvest task and automatically queue discovered PDFs for Chandra.
Returns HarvestResult with persons/relationships already ingested to graph.
"""
browser = Browser(config=get_browser_config())
agent = Agent(task=task, llm=llm, browser=browser)
raw_result = await agent.run()
await browser.close()
result = parse_harvest_result(raw_result.final_result(), source_url)
# 1. Ingest persons/relationships to graph DB (see 3.2)
ingest_harvest_result(result, db, resolver)
# 2. Download and queue PDFs for Chandra
if result.pdf_links:
downloaded = []
for pdf_url in result.pdf_links:
try:
path = await download_pdf(pdf_url, PDF_DOWNLOAD_DIR)
downloaded.append(path)
except Exception as e:
result.errors.append(f"PDF download failed: {pdf_url} — {e}")
if downloaded:
queue_for_chandra(downloaded, chandra_queue)
return result
GPU serialization note: The Chandra pipeline must not run while any MLX/MPS job is active (see docs/pillar-pdf-ingestion-notes.md). The Chandra queue processor checks for active MLX jobs before starting:
import subprocess
def chandra_queue_is_safe_to_run() -> bool:
"""True only if no MLX/MPS job is currently running."""
result = subprocess.run(
["pgrep", "-f", "mlx|chandra|whisper|scrape_slides|ingest_pdfs"],
capture_output=True, text=True
)
return result.returncode != 0 # pgrep returns 1 if no matches
3.2 Browser-Use JSON → SQLite Graph Ingest
The HarvestResult dataclass maps directly to the SQLite schema. The ingest function handles node upsert, entity resolution, and edge creation:
from sqlite_graph import (
GraphDB, Source, Node, Edge, EntityResolver
)
from datetime import datetime
def ingest_harvest_result(
result: "HarvestResult",
db: GraphDB,
resolver: EntityResolver
) -> dict:
"""
Ingest a HarvestResult into the SQLite graph DB.
Returns dict of created/matched IDs for logging.
"""
stats = {"persons_created": 0, "persons_matched": 0,
"orgs_created": 0, "edges_created": 0, "errors": []}
# --- Register source ---
source = Source(
source_type="web_scrape",
url=result.source_url,
filing_date=result.harvest_timestamp[:10],
reliability_score=0.72, # web scrape default
retrieved_at=result.harvest_timestamp
)
source_id = db.upsert_source(source)
# --- Upsert organization node ---
org_node_id = None
if result.organization:
org = result.organization
org_node = Node(
name=org.name,
node_type=org.node_type,
node_subtype=org.node_subtype,
canonical_id=org.canonical_id,
metadata={"website": org.url, "source_url": org.source_url},
source_id=source_id
)
existing = resolver.resolve_node(org_node)
if existing:
org_node_id = existing
stats["orgs_created"] += 0 # matched
else:
org_node_id = db.upsert_node(org_node)
stats["orgs_created"] += 1
# --- Upsert person nodes ---
person_id_map: dict[str, str] = {} # name -> node_id
for person in result.persons:
p_node = Node(
name=person.name,
node_type="person",
node_subtype=person.node_subtype,
metadata={
k: v for k, v in {
"title": person.title,
"email": person.email,
"phone": person.phone,
"bio_url": person.bio_url,
"source_url": person.source_url,
"organization": person.organization
}.items() if v
},
source_id=source_id
)
existing = resolver.resolve_node(p_node)
if existing:
person_id_map[person.name] = existing
stats["persons_matched"] += 1
else:
pid = db.upsert_node(p_node)
person_id_map[person.name] = pid
stats["persons_created"] += 1
# --- Upsert edges ---
today = datetime.utcnow().date().isoformat()
for rel in result.relationships:
person_id = person_id_map.get(rel.person_name)
if not person_id:
stats["errors"].append(f"No person_id for {rel.person_name}")
continue
# Resolve org_node_id for this relationship
target_org_id = org_node_id
if not target_org_id and rel.organization_name:
stub = Node(name=rel.organization_name, node_type="firm")
existing_org = resolver.resolve_node(stub)
target_org_id = existing_org if existing_org else db.upsert_node(stub)
if not target_org_id:
stats["errors"].append(f"No org_id for {rel.organization_name}")
continue
edge = Edge(
from_node_id=person_id,
to_node_id=target_org_id,
edge_type=rel.edge_type,
weight=rel.confidence,
valid_from=rel.valid_from or today,
metadata={"title": rel.title} if rel.title else {},
source_id=source_id
)
db.upsert_edge(edge)
stats["edges_created"] += 1
return stats
3.3 Deduplication Against Existing Graph Nodes
Before inserting harvested persons, the EntityResolver runs against the existing graph. The resolver uses:
- Exact canonical_id match — if the harvest extracted an EIN or CRD, this resolves immediately.
- Normalized name + organization match — strips titles, lowercases, matches against existing nodes with the same employer.
- Levenshtein distance ≤ 2 — near-miss names (e.g., “William J. Chen” vs “William Chen”) queued for review.
The critical path for web-harvested names is the name+org combination because these names have no EIN/CRD. For harvested persons, supplement the resolver with the organization context:
def resolve_person_with_org_context(
resolver: EntityResolver,
person_name: str,
org_name: str,
db: GraphDB
) -> str | None:
"""
Enhanced resolution: match person name + look for existing edge to same org.
Returns existing node_id if confident match, else None.
"""
conn = db.connect()
norm_name = normalize_name(person_name)
# Find all existing person nodes with similar normalized names
# Use FTS5 for approximate name search
fts_results = db.search_nodes(norm_name, node_type="person", limit=20)
for candidate in fts_results:
score, method = candidate_match_score(
person_name, candidate["name"],
None, candidate.get("canonical_id")
)
if score < 0.70:
continue
# Check if candidate has any WORKS_AT/TRUSTEE_OF/COMMITTEE_MEMBER_OF
# edge to an organization with a similar name to org_name
active_edges = db.get_active_edges(
candidate["id"],
edge_type=None # Check all employment-type edges
)
for edge in active_edges:
if edge["edge_type"] not in ("WORKS_AT", "TRUSTEE_OF",
"COMMITTEE_MEMBER_OF", "FORMERLY_AT"):
continue
org_node = db.get_node(edge["to_node_id"])
if not org_node:
continue
org_score, _ = candidate_match_score(
org_name, org_node["name"], None, org_node.get("canonical_id")
)
if org_score >= 0.75 and score >= 0.80:
return candidate["id"] # High confidence: same name + same org
return None # No match; create new node
3.4 Change Detection and Job-Change Edges
On each scheduled re-harvest of a target page, the pipeline diffs new results against existing graph state to detect personnel changes:
from sqlite_graph import ingest_job_change
def detect_and_record_changes(
new_result: "HarvestResult",
org_node_id: str,
db: GraphDB,
resolver: EntityResolver,
source_id: str
):
"""
Compare new harvest against existing active edges for this organization.
- People who appear in new_result but not in current graph → new TRUSTEE_OF/WORKS_AT edge
- People who were in graph but absent from new_result → close the edge (valid_to = today)
- People present in both with different titles → close old edge, open new one
"""
conn = db.connect()
today = datetime.utcnow().date().isoformat()
# Current active persons at this org (all relevant edge types)
current_rows = conn.execute("""
SELECT p.id, p.name, e.id AS edge_id, e.edge_type,
json_extract(e.metadata, '$.title') AS title
FROM edges e
JOIN nodes p ON p.id = e.from_node_id AND p.node_type = 'person'
WHERE e.to_node_id = ?
AND e.edge_type IN ('TRUSTEE_OF', 'COMMITTEE_MEMBER_OF', 'WORKS_AT')
AND e.valid_to IS NULL
""", (org_node_id,)).fetchall()
current_by_name = {normalize_name(r["name"]): dict(r) for r in current_rows}
new_names = {normalize_name(p.name): p for p in new_result.persons}
# People removed (in current, not in new)
for norm_name, current_row in current_by_name.items():
if norm_name not in new_names:
# Close the edge — person no longer listed
db.close_edge(current_row["edge_id"], valid_to=today)
# Optionally: add a note to resolution_queue for human review
# before assuming departure vs. page structure change
# People added (in new, not in current)
for norm_name, new_person in new_names.items():
if norm_name not in current_by_name:
# New person — already ingested by ingest_harvest_result;
# the edge was created there. No action needed here.
pass
else:
# Check for title change
existing_title = current_by_name[norm_name].get("title")
if new_person.title and new_person.title != existing_title:
# Close old edge, let ingest_harvest_result create the new one
db.close_edge(current_by_name[norm_name]["edge_id"], valid_to=today)
3.5 Error Handling and Manual Review Queue
Three error categories require different handling:
Site structure change — The LLM fails to find the expected content (returns empty persons list or explicitly states “page structure not recognized”). Flag for manual review:
REVIEW_QUEUE_PATH = Path("research/19-agent-frameworks/raw/browser-harvest/review_queue.jsonl")
def flag_for_review(url: str, reason: str, raw_output: str):
"""Write a review item to the JSONL queue."""
import json
item = {
"url": url,
"reason": reason,
"raw_output": raw_output[:500], # truncate for storage
"flagged_at": datetime.utcnow().isoformat() + "Z",
"status": "pending"
}
with REVIEW_QUEUE_PATH.open("a") as f:
f.write(json.dumps(item) + "\n")
CAPTCHA encounter — browser-use will report the page contains a CAPTCHA challenge. The task specification should include: “If you encounter a CAPTCHA, stop and return an error result rather than attempting to solve it.” The failed URL goes to the review queue for a manual or headful retry.
Blocked request (HTTP 403/429) — implement exponential backoff with jitter for 429; treat 403 as a hard block and add to the review queue:
import asyncio, random
async def resilient_harvest(
task: str, llm, max_retries: int = 3
) -> dict:
"""Run harvest with retry on rate limit errors."""
for attempt in range(max_retries):
try:
browser = Browser(config=get_browser_config())
agent = Agent(task=task, llm=llm, browser=browser)
result = await agent.run()
await browser.close()
return result.final_result()
except Exception as e:
if "429" in str(e) or "rate limit" in str(e).lower():
wait = (2 ** attempt) * 30 + random.uniform(0, 10)
await asyncio.sleep(wait)
else:
raise
raise RuntimeError(f"Max retries exceeded for harvest task")
Part 4 — Concrete Task Examples
4.1 CalPERS Board Member Roster → Person Nodes
"""
harvest_calpers_board.py — Extract CalPERS board member roster and create
TRUSTEE_OF edges from each person to the CalPERS firm node.
"""
import asyncio
import json
from datetime import datetime
from langchain_anthropic import ChatAnthropic
from browser_use import Agent, Browser, BrowserConfig
from sqlite_graph import GraphDB, EntityResolver
CALPERS_BOARD_URL = "https://www.calpers.ca.gov/about/organization/board-of-administration"
CALPERS_EIN = "EIN:946001385"
CALPERS_TASK = """
Navigate to https://www.calpers.ca.gov/about/organization/board-of-administration
Your goal is to extract the complete current board of administration member roster.
For each board member, extract:
- Full name (as displayed)
- Title or role designation (e.g., "Board President", "State Controller", "Public Member")
- The basis of their seat if stated (e.g., "Appointed by Governor", "Elected by State employees", "Ex officio")
- Their term expiration date if listed
- A link to their individual bio page if one exists
Also look for any links to annual reports or investment documents on this page or in the navigation.
Collect all PDF links that appear to be annual reports, investment reports, or board meeting materials.
Return the results as valid JSON matching exactly this schema:
{
"organization": {
"name": "California Public Employees Retirement System",
"canonical_id": "EIN:946001385",
"url": "https://www.calpers.ca.gov",
"node_type": "firm",
"node_subtype": "public_pension"
},
"persons": [
{
"name": "string",
"title": "string",
"node_type": "person",
"node_subtype": "trustee",
"bio_url": "string or null",
"source_url": "https://www.calpers.ca.gov/about/organization/board-of-administration"
}
],
"relationships": [
{
"person_name": "string",
"organization_name": "California Public Employees Retirement System",
"edge_type": "TRUSTEE_OF",
"title": "string",
"confidence": 0.90
}
],
"pdf_links": ["string array of PDF URLs found on page"],
"errors": []
}
If a board member's name is unclear, include what you can see and flag it in errors.
Do not invent information. Only include what is explicitly on the page.
"""
async def harvest_calpers_board(db: GraphDB, resolver: EntityResolver) -> dict:
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0, max_tokens=4096)
browser = Browser(config=BrowserConfig(
headless=True,
slow_mo=1500,
extra_chromium_args=["--disable-blink-features=AutomationControlled"]
))
agent = Agent(task=CALPERS_TASK, llm=llm, browser=browser)
raw = await agent.run()
await browser.close()
raw_json = raw.final_result()
if isinstance(raw_json, str):
# LLM may return JSON as a string
try:
parsed = json.loads(raw_json)
except json.JSONDecodeError:
# Extract JSON block from markdown code fence if present
import re
match = re.search(r'```json\s*([\s\S]+?)\s*```', raw_json)
parsed = json.loads(match.group(1)) if match else {}
else:
parsed = raw_json
result = HarvestResult(
source_url=CALPERS_BOARD_URL,
harvest_timestamp=datetime.utcnow().isoformat() + "Z",
organization=HarvestOrganization(**parsed.get("organization", {})),
persons=[HarvestPerson(**p) for p in parsed.get("persons", [])],
relationships=[HarvestRelationship(**r) for r in parsed.get("relationships", [])],
pdf_links=parsed.get("pdf_links", []),
errors=parsed.get("errors", []),
raw_llm_output=str(raw_json)
)
stats = ingest_harvest_result(result, db, resolver)
print(f"CalPERS harvest: {stats['persons_created']} new, "
f"{stats['persons_matched']} matched, {stats['edges_created']} edges, "
f"{len(result.pdf_links)} PDFs queued")
return stats
if __name__ == "__main__":
db = GraphDB("investors.db")
db.initialize_schema()
resolver = EntityResolver(db)
asyncio.run(harvest_calpers_board(db, resolver))
4.2 University Endowment IC Page → Person Nodes with COMMITTEE_MEMBER_OF Edges
"""
harvest_endowment_ic.py — Extract university endowment investment committee
member names and create COMMITTEE_MEMBER_OF edges.
Works for DUMAC (Duke), PRINCO (Princeton), MITIMCO (MIT) which publish
IC member names. Adapt the URL and org metadata per target.
"""
import asyncio
import json
from langchain_anthropic import ChatAnthropic
from browser_use import Agent, Browser, BrowserConfig
ENDOWMENT_TARGETS = [
{
"url": "https://dumac.org/about",
"org_name": "Duke Management Company",
"org_canonical": None, # No EIN for DUMAC directly
"org_subtype": "endowment",
"edge_type": "COMMITTEE_MEMBER_OF"
},
{
"url": "https://www.princetoninvestment.com/about/investment-committee",
"org_name": "Princeton University Investment Company",
"org_canonical": None,
"org_subtype": "endowment",
"edge_type": "COMMITTEE_MEMBER_OF"
},
{
"url": "https://mitimco.org/about/investment-committee",
"org_name": "MIT Investment Management Company",
"org_canonical": None,
"org_subtype": "endowment",
"edge_type": "COMMITTEE_MEMBER_OF"
},
]
def build_endowment_task(target: dict) -> str:
return f"""
Navigate to {target['url']}
Extract all named investment committee members, board members, or advisory board
members listed on this page for {target['org_name']}.
For each person found, extract:
- Full name (exactly as shown)
- Title or role (e.g., "Chair, Investment Committee", "External Member", "Trustee")
- Professional affiliation if listed (e.g., their day job firm, if mentioned)
- Link to their bio if available on the page
Also collect any annual report PDF links, endowment annual report links,
or investment committee charter documents linked from this page.
If the page does not list individual committee members by name, try navigating
to linked sub-pages such as /leadership, /governance, /investment-committee, or /people.
Report what you found and what you tried.
Return as JSON:
{{
"organization": {{
"name": "{target['org_name']}",
"canonical_id": null,
"url": "{target['url']}",
"node_type": "firm",
"node_subtype": "{target['org_subtype']}"
}},
"persons": [
{{
"name": "string",
"title": "string",
"organization": "their primary employer if listed or null",
"bio_url": "string or null",
"node_type": "person",
"node_subtype": "committee_member",
"source_url": "{target['url']}"
}}
],
"relationships": [
{{
"person_name": "string",
"organization_name": "{target['org_name']}",
"edge_type": "{target['edge_type']}",
"title": "their committee title",
"confidence": 0.88
}}
],
"pdf_links": [],
"errors": []
}}
"""
async def harvest_endowment_ic_pages(db: GraphDB, resolver: EntityResolver):
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0, max_tokens=4096)
for target in ENDOWMENT_TARGETS:
task = build_endowment_task(target)
browser = Browser(config=BrowserConfig(headless=True, slow_mo=2000))
agent = Agent(task=task, llm=llm, browser=browser)
try:
raw = await agent.run()
await browser.close()
result_json = raw.final_result()
if isinstance(result_json, str):
import re
match = re.search(r'```json\s*([\s\S]+?)\s*```', result_json)
parsed = json.loads(match.group(1) if match else result_json)
else:
parsed = result_json
result = HarvestResult(
source_url=target["url"],
harvest_timestamp=datetime.utcnow().isoformat() + "Z",
organization=HarvestOrganization(**parsed.get("organization", {})),
persons=[HarvestPerson(**p) for p in parsed.get("persons", [])],
relationships=[HarvestRelationship(**r) for r in parsed.get("relationships", [])],
pdf_links=parsed.get("pdf_links", []),
errors=parsed.get("errors", [])
)
stats = ingest_harvest_result(result, db, resolver)
print(f"{target['org_name']}: {stats}")
except Exception as e:
flag_for_review(target["url"], str(e), "")
await browser.close()
# Polite delay between targets
await asyncio.sleep(5)
4.3 NEPC Staff Directory → Person Nodes with WORKS_AT Edges
"""
harvest_nepc_staff.py — Extract NEPC investment consultant staff directory.
Handles JavaScript-rendered team pages.
"""
import asyncio
import json
from langchain_anthropic import ChatAnthropic
from browser_use import Agent, Browser, BrowserConfig, Controller
NEPC_TASK = """
Navigate to https://nepc.com/team
This page lists NEPC's investment consulting staff. It may require scrolling
or clicking filters to see the full team list.
First, look for any filter controls that expand the visible team (e.g., "All",
"Investment Consulting", "Research"). Click to show all staff if a filter exists.
Scroll down to load all team members (the page may use lazy loading).
For each team member listed, extract:
- Full name
- Title / position
- Practice area or specialty if listed (e.g., "Defined Benefit", "Endowments & Foundations")
- Geographic office if listed
- Link to their individual profile page if one exists
Return as JSON:
{
"organization": {
"name": "NEPC LLC",
"canonical_id": null,
"url": "https://nepc.com",
"node_type": "firm",
"node_subtype": "advisory_firm"
},
"persons": [
{
"name": "string",
"title": "string",
"organization": "NEPC LLC",
"bio_url": "string or null",
"node_type": "person",
"node_subtype": "consultant",
"source_url": "https://nepc.com/team"
}
],
"relationships": [
{
"person_name": "string",
"organization_name": "NEPC LLC",
"edge_type": "WORKS_AT",
"title": "their title",
"confidence": 0.82
}
],
"pdf_links": [],
"errors": []
}
If the page blocks automated access or shows a CAPTCHA, report this in errors
rather than attempting to bypass it.
"""
async def harvest_nepc_staff(db: GraphDB, resolver: EntityResolver) -> dict:
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0, max_tokens=8096)
# Use residential proxy for consultant firm sites
browser = Browser(config=BrowserConfig(
headless=True,
slow_mo=3000,
extra_chromium_args=[
"--disable-blink-features=AutomationControlled",
"--window-size=1920,1080"
]
# Add proxy here if encountering blocks:
# proxy={"server": "...", "username": "...", "password": "..."}
))
agent = Agent(task=NEPC_TASK, llm=llm, browser=browser)
raw = await agent.run()
await browser.close()
result_json = raw.final_result()
if isinstance(result_json, str):
import re
match = re.search(r'```json\s*([\s\S]+?)\s*```', result_json)
parsed = json.loads(match.group(1) if match else result_json)
else:
parsed = result_json
result = HarvestResult(
source_url="https://nepc.com/team",
harvest_timestamp=datetime.utcnow().isoformat() + "Z",
organization=HarvestOrganization(**parsed.get("organization", {})),
persons=[HarvestPerson(**p) for p in parsed.get("persons", [])],
relationships=[HarvestRelationship(**r) for r in parsed.get("relationships", [])],
pdf_links=[],
errors=parsed.get("errors", [])
)
# For consultant staff, cross-reference with Form ADV Part 2B
# to enrich with employment histories
stats = ingest_harvest_result(result, db, resolver)
print(f"NEPC harvest: {stats}")
return stats
4.4 Following PDF Links from a Pension Annual Report Page → Chandra Queue
"""
harvest_pension_annual_report_pdfs.py — Navigate a pension fund website,
discover annual report PDFs and board meeting minute PDFs,
download them, and queue for Chandra OCR.
This task does NOT extract names directly — it is a PDF discovery task.
Chandra + spaCy NER extract names from the PDFs after ingestion.
"""
import asyncio
from pathlib import Path
from langchain_anthropic import ChatAnthropic
from browser_use import Agent, Browser, BrowserConfig
PDF_QUEUE = Path("research/19-agent-frameworks/raw/browser-harvest/chandra_queue.tsv")
PDF_DOWNLOAD_DIR = Path("research/19-agent-frameworks/raw/browser-harvest/pdfs")
PDF_DISCOVERY_TASK = """
Navigate to {base_url}
Your goal is to find and collect URLs for downloadable PDF documents that are:
1. Annual investment reports or annual reports
2. Board meeting agendas or board meeting minutes
3. Investment schedule or investment portfolio disclosures
4. Investment policy statements
Look in: navigation menus (Publications, Reports, Documents, Board, Meetings),
footer links, and any "Download" or "PDF" buttons visible on the page.
For each PDF found, record:
- The full URL of the PDF
- A description of what the document appears to be based on its link text or surrounding context
- The approximate year/date if determinable from the link text
Return as JSON:
{{
"source_url": "{base_url}",
"pdf_links": [
{{
"url": "string (full URL)",
"description": "string",
"year": "string or null"
}}
],
"errors": []
}}
Navigate up to 3 levels deep following links that appear to lead to document repositories.
Do not navigate to external sites (links leaving {domain}).
"""
PENSION_PDF_TARGETS = [
{"name": "CalPERS", "base_url": "https://www.calpers.ca.gov/investments",
"domain": "calpers.ca.gov"},
{"name": "CalSTRS", "base_url": "https://www.calstrs.com/investment-reports",
"domain": "calstrs.com"},
{"name": "WSIB", "base_url": "https://www.sib.wa.gov/investments",
"domain": "sib.wa.gov"},
{"name": "Texas TRS", "base_url": "https://www.trs.texas.gov/investments",
"domain": "trs.texas.gov"},
{"name": "TRSIL", "base_url": "https://www.trs.illinois.gov/investments",
"domain": "trs.illinois.gov"},
]
async def harvest_pension_pdfs():
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0, max_tokens=4096)
PDF_DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
all_discovered = []
for target in PENSION_PDF_TARGETS:
task = PDF_DISCOVERY_TASK.format(
base_url=target["base_url"],
domain=target["domain"]
)
browser = Browser(config=BrowserConfig(headless=True, slow_mo=2000))
agent = Agent(task=task, llm=llm, browser=browser)
try:
raw = await agent.run()
await browser.close()
import json, re
result_str = raw.final_result()
if isinstance(result_str, str):
match = re.search(r'```json\s*([\s\S]+?)\s*```', result_str)
parsed = json.loads(match.group(1) if match else result_str)
else:
parsed = result_str
pdf_entries = parsed.get("pdf_links", [])
print(f"{target['name']}: discovered {len(pdf_entries)} PDFs")
for entry in pdf_entries:
url = entry.get("url", "")
if not url or not url.lower().endswith(".pdf"):
continue
try:
path = await download_pdf(url, PDF_DOWNLOAD_DIR)
all_discovered.append({
"pension": target["name"],
"path": str(path),
"url": url,
"description": entry.get("description", ""),
"year": entry.get("year")
})
except Exception as e:
print(f" Download failed: {url} — {e}")
except Exception as e:
flag_for_review(target["base_url"], str(e), "")
await asyncio.sleep(5)
# Write queue for Chandra
with PDF_QUEUE.open("w") as f:
f.write("path\turl\tdescription\tyear\tpension\ttimestamp\n")
for item in all_discovered:
ts = datetime.utcnow().isoformat() + "Z"
f.write(f"{item['path']}\t{item['url']}\t"
f"{item['description']}\t{item.get('year','')}\t"
f"{item['pension']}\t{ts}\n")
print(f"\nTotal PDFs queued for Chandra: {len(all_discovered)}")
print(f"Queue file: {PDF_QUEUE}")
return all_discovered
4.5 Conference Speaker Bios from P&I Event Page → Person Nodes
"""
harvest_pi_conference_speakers.py — Extract speaker bios from a P&I
conference page, creating person nodes with institutional affiliations.
Speaker bios at P&I events are a high-quality, self-reported source:
the person chose to present, their bio is written for professional context,
and their current title + organization are explicitly stated.
"""
import asyncio
import json
import re
from langchain_anthropic import ChatAnthropic
from browser_use import Agent, Browser, BrowserConfig
PI_SPEAKER_TASK = """
Navigate to {conference_url}
Your goal is to extract speaker bios from this Pensions & Investments conference page.
Look for a "Speakers" section, speaker listing, or agenda with speaker bios.
You may need to navigate to a sub-page (e.g., /speakers or /agenda).
For each speaker, extract:
- Full name
- Current title (exactly as stated in bio)
- Current organization / employer (exactly as stated)
- Any investment committee memberships mentioned (e.g., "serves on the investment committee of...")
- Any trustee roles mentioned
- Whether they are listed as an LP (pension fund, endowment, foundation) vs GP (fund manager) vs consultant
- Their session topic or panel title if available
Return as JSON:
{{
"conference": {{
"name": "{conference_name}",
"url": "{conference_url}",
"date": "approximate date if shown"
}},
"speakers": [
{{
"name": "string",
"title": "string",
"organization": "string",
"speaker_type": "LP|GP|consultant|other",
"committee_memberships": ["list of orgs they serve on IC of"],
"trustee_roles": ["list of orgs they are trustee of"],
"session_topic": "string or null",
"bio_url": "link to full bio if separate page exists"
}}
],
"errors": []
}}
If speaker bios are on individual sub-pages, navigate to the first 10 bios
and extract those. Report how many total speakers are listed.
"""
async def harvest_pi_conference_speakers(
conference_url: str,
conference_name: str,
db: GraphDB,
resolver: EntityResolver
) -> dict:
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0, max_tokens=8096)
task = PI_SPEAKER_TASK.format(
conference_url=conference_url,
conference_name=conference_name
)
browser = Browser(config=BrowserConfig(headless=True, slow_mo=2000))
agent = Agent(task=task, llm=llm, browser=browser)
raw = await agent.run()
await browser.close()
result_str = raw.final_result()
if isinstance(result_str, str):
match = re.search(r'```json\s*([\s\S]+?)\s*```', result_str)
parsed = json.loads(match.group(1) if match else result_str)
else:
parsed = result_str
stats = {"persons_created": 0, "edges_created": 0, "committee_edges": 0}
source = Source(
source_type="web_scrape",
url=conference_url,
reliability_score=0.80, # Speaker self-reports are high quality
retrieved_at=datetime.utcnow().isoformat() + "Z"
)
source_id = db.upsert_source(source)
today = datetime.utcnow().date().isoformat()
for speaker in parsed.get("speakers", []):
# Person node
p_node = Node(
name=speaker["name"],
node_type="person",
node_subtype="trustee" if speaker.get("trustee_roles") else "staff",
metadata={
"title": speaker.get("title"),
"speaker_type": speaker.get("speaker_type"),
"session_topic": speaker.get("session_topic"),
"source_url": conference_url
},
source_id=source_id
)
existing = resolver.resolve_node(p_node)
person_id = existing if existing else db.upsert_node(p_node)
if not existing:
stats["persons_created"] += 1
# Primary employer WORKS_AT edge
if speaker.get("organization"):
org_stub = Node(
name=speaker["organization"],
node_type="firm",
source_id=source_id
)
existing_org = resolver.resolve_node(org_stub)
org_id = existing_org if existing_org else db.upsert_node(org_stub)
# Determine edge type based on speaker role and title
title_lower = (speaker.get("title") or "").lower()
if any(x in title_lower for x in ["trustee", "board member", "board chair"]):
primary_edge_type = "TRUSTEE_OF"
elif any(x in title_lower for x in ["investment committee", "ic member", "committee member"]):
primary_edge_type = "COMMITTEE_MEMBER_OF"
else:
primary_edge_type = "WORKS_AT"
edge = Edge(
from_node_id=person_id,
to_node_id=org_id,
edge_type=primary_edge_type,
weight=0.82,
valid_from=today,
metadata={"title": speaker.get("title"),
"conference": conference_name},
source_id=source_id
)
db.upsert_edge(edge)
stats["edges_created"] += 1
# Additional committee memberships from bio text
for committee_org_name in speaker.get("committee_memberships", []):
if not committee_org_name:
continue
committee_org = Node(
name=committee_org_name,
node_type="firm",
source_id=source_id
)
existing_c = resolver.resolve_node(committee_org)
committee_id = existing_c if existing_c else db.upsert_node(committee_org)
committee_edge = Edge(
from_node_id=person_id,
to_node_id=committee_id,
edge_type="COMMITTEE_MEMBER_OF",
weight=0.78, # Slight discount — extracted from bio text, not explicit roster
valid_from=today,
metadata={"source": "conference_bio", "conference": conference_name},
source_id=source_id
)
db.upsert_edge(committee_edge)
stats["committee_edges"] += 1
# Trustee roles from bio text
for trustee_org_name in speaker.get("trustee_roles", []):
if not trustee_org_name:
continue
trustee_org = Node(
name=trustee_org_name,
node_type="firm",
source_id=source_id
)
existing_t = resolver.resolve_node(trustee_org)
trustee_org_id = existing_t if existing_t else db.upsert_node(trustee_org)
trustee_edge = Edge(
from_node_id=person_id,
to_node_id=trustee_org_id,
edge_type="TRUSTEE_OF",
weight=0.78,
valid_from=today,
metadata={"source": "conference_bio", "conference": conference_name},
source_id=source_id
)
db.upsert_edge(trustee_edge)
stats["committee_edges"] += 1
print(f"Conference harvest ({conference_name}): {stats}")
return stats
Part 5 — Operational Pipeline Design
5.1 Scheduler: APScheduler
APScheduler is the right choice for this pipeline: it runs in-process (no separate broker required), supports cron-style and interval triggers, persists job state to SQLite (the same DB we already have), and integrates cleanly with asyncio.
"""
harvest_scheduler.py — APScheduler configuration for recurring harvests.
Run with: python harvest_scheduler.py
Persists job state to investors.db (same SQLite file as graph).
"""
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.asyncio import AsyncIOExecutor
from datetime import datetime
from sqlite_graph import GraphDB, EntityResolver
from harvest_calpers_board import harvest_calpers_board
from harvest_endowment_ic import harvest_endowment_ic_pages
from harvest_nepc_staff import harvest_nepc_staff
from harvest_pension_annual_report_pdfs import harvest_pension_pdfs
from harvest_pi_conference_speakers import harvest_pi_conference_speakers
DB_PATH = "investors.db"
# APScheduler config
jobstores = {
"default": SQLAlchemyJobStore(url=f"sqlite:///{DB_PATH}",
tablename="apscheduler_jobs")
}
executors = {"default": AsyncIOExecutor()}
job_defaults = {"coalesce": True, "max_instances": 1, "misfire_grace_time": 3600}
def get_db_and_resolver():
db = GraphDB(DB_PATH)
db.initialize_schema()
resolver = EntityResolver(db, levenshtein_threshold=2, min_confidence=0.60)
return db, resolver
async def run_calpers_harvest():
db, resolver = get_db_and_resolver()
await harvest_calpers_board(db, resolver)
async def run_endowment_harvest():
db, resolver = get_db_and_resolver()
await harvest_endowment_ic_pages(db, resolver)
async def run_consultant_harvest():
db, resolver = get_db_and_resolver()
await harvest_nepc_staff(db, resolver)
async def run_pdf_discovery():
await harvest_pension_pdfs()
async def run_pi_conference_harvest():
db, resolver = get_db_and_resolver()
# Current P&I conference schedule — update annually
conferences = [
{
"url": "https://www.pionline.com/events/investment-consulting",
"name": "P&I Investment Consulting Conference 2026"
},
{
"url": "https://www.pionline.com/events/alternative-investments",
"name": "P&I Alternative Investments Conference 2026"
}
]
for conf in conferences:
await harvest_pi_conference_speakers(
conf["url"], conf["name"], db, resolver
)
await asyncio.sleep(5)
def main():
scheduler = AsyncIOScheduler(
jobstores=jobstores,
executors=executors,
job_defaults=job_defaults
)
# Tier 1 — Weekly harvests (fast-changing or high-value)
# Public pension board rosters: monthly is sufficient; weekly for CalPERS/CalSTRS
# given their size and frequency of trustee news
scheduler.add_job(
run_calpers_harvest, "cron",
day_of_week="mon", hour=2, minute=0,
id="calpers_weekly", replace_existing=True,
name="CalPERS board roster harvest"
)
# Tier 1 — Monthly harvests
scheduler.add_job(
run_endowment_harvest, "cron",
day=1, hour=3, minute=0,
id="endowment_monthly", replace_existing=True,
name="University endowment IC pages"
)
scheduler.add_job(
run_pdf_discovery, "cron",
day=5, hour=4, minute=0,
id="pdf_discovery_monthly", replace_existing=True,
name="Pension annual report PDF discovery"
)
# Tier 2 — Quarterly harvests (consultant directories change slowly)
scheduler.add_job(
run_consultant_harvest, "cron",
month="1,4,7,10", day=10, hour=5, minute=0,
id="consultant_quarterly", replace_existing=True,
name="Investment consultant staff directories"
)
# Conference speaker harvest — event-triggered, not cron
# Run manually: asyncio.run(run_pi_conference_harvest())
# Or schedule around known conference dates
scheduler.start()
print(f"Harvest scheduler started at {datetime.now()}")
print("Jobs scheduled:")
for job in scheduler.get_jobs():
print(f" {job.id}: next run {job.next_run_time}")
try:
asyncio.get_event_loop().run_forever()
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
print("Scheduler stopped.")
if __name__ == "__main__":
main()
5.2 Priority Queue — Re-harvest Cadence by Site
| Tier | Site Category | Harvest Cadence | Rationale |
|---|---|---|---|
| 1A | CalPERS, CalSTRS, NYCDRS | Weekly | High AUM, frequent trustee news, anchor of graph |
| 1B | Texas TRS, WSIB, LACERA, Ohio STRS, TRSIL | Monthly | Moderate change frequency |
| 1C | Other large public pensions ($10B+) | Monthly | Standard public pension cadence |
| 1D | University endowment IC pages (active disclosers) | Quarterly | IC composition changes slowly |
| 1E | Foundation board/IC pages | Quarterly | Annual board elections, slow churn |
| 1F | EDGAR Form ADV new filings | Weekly | EDGAR RSS feed for new ADV filers |
| 2A | Conference speaker pages (upcoming) | 2 weeks pre-conference + 4 weeks post | |
| 2B | Consultant staff directories (NEPC, Callan, RVK) | Quarterly | |
| 2C | Consultant staff directories (Mercer, Aon, WTW) | Semi-annually | Larger firms, slower staff change visibility |
| 2D | Form ADV Part 2B bulk download → Chandra | Quarterly | Large-scale supervised person extraction |
| 3A | Google-dork LinkedIn profiles (targeted enrichment) | As-needed | Triggered by specific graph gap-filling requests |
5.3 Cost Estimate
browser-use API costs (Claude Sonnet 4.6 backend):
Typical harvest task: ~5,000–15,000 input tokens (page content) + 500–2,000 output tokens (JSON result).
| Harvest type | Avg tokens | Cost per run | Monthly cost (est.) |
|---|---|---|---|
| Pension board roster (simple page) | 8K in / 1K out | ~$0.05 | $0.20 (weekly) |
| Endowment IC page (complex navigation) | 20K in / 2K out | ~$0.13 | $0.52 (quarterly) |
| Consultant staff directory (JS-heavy) | 30K in / 3K out | ~$0.19 | $0.19 (quarterly) |
| Conference speaker page (multiple bios) | 40K in / 4K out | ~$0.27 | $0.54 (2x/conf x 4 confs) |
| PDF discovery sweep (5 pensions) | 15K in / 1.5K out | ~$0.10 | $0.10 (monthly) |
Total estimated LLM cost: under $5/month for the full Tier 1 + Tier 2 harvest schedule at Claude Sonnet 4.6 pricing. This is the primary operating cost; Playwright itself is free and self-hosted.
Residential proxy costs (for Tier 2 consultant sites): Oxylabs residential proxy pool runs ~$15/GB. Typical consultant directory harvest: 5–15MB page data. Total proxy cost: under $2/month.
Self-hosted Playwright vs. browser-use Cloud:
browser-use offers a cloud execution environment (cloud.browser-use.com) that eliminates local browser management. Pricing as of mid-2026: $0.20–$0.50 per task execution depending on complexity. For the harvest schedule above (~50 task executions/month), cloud cost would be $10–$25/month — marginally higher than self-hosted but eliminates infrastructure maintenance.
Recommendation for this project: Self-hosted Playwright on the existing Mac Studio (same machine as Chandra/MLX). The harvest tasks are short-running and do not conflict with GPU workloads (browser automation is CPU/network bound; Chandra is GPU bound). Schedule harvests to complete before Chandra queue processing begins.
5.4 Monitoring: Success/Failure Rates and Coverage Metrics
"""
harvest_monitor.py — Coverage and health metrics for the harvest pipeline.
Run weekly to review graph growth and identify stale data.
"""
import sqlite3
from datetime import datetime, timedelta
DB_PATH = "investors.db"
def harvest_coverage_report(db_path: str = DB_PATH) -> dict:
"""
Returns a dict of coverage metrics for the investor graph.
Run weekly to monitor pipeline health.
"""
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
metrics = {}
# Total node counts by type
metrics["node_counts"] = dict(conn.execute("""
SELECT node_type, COUNT(*) as count
FROM nodes GROUP BY node_type
""").fetchall())
# Node counts by subtype
metrics["firm_subtypes"] = dict(conn.execute("""
SELECT node_subtype, COUNT(*) as count
FROM nodes WHERE node_type = 'firm'
GROUP BY node_subtype
""").fetchall())
# Active edge counts by type
metrics["active_edge_counts"] = dict(conn.execute("""
SELECT edge_type, COUNT(*) as count
FROM edges WHERE valid_to IS NULL
GROUP BY edge_type
""").fetchall())
# Harvest run stats by source type (last 30 days)
thirty_days_ago = (datetime.utcnow() - timedelta(days=30)).isoformat()
metrics["recent_harvests"] = [dict(r) for r in conn.execute("""
SELECT source_type,
COUNT(*) as harvest_count,
COUNT(DISTINCT url) as unique_urls
FROM sources
WHERE retrieved_at >= ?
GROUP BY source_type
ORDER BY harvest_count DESC
""", (thirty_days_ago,)).fetchall()]
# New nodes per week (last 8 weeks)
metrics["new_nodes_weekly"] = [dict(r) for r in conn.execute("""
SELECT strftime('%Y-W%W', created_at) as week,
COUNT(*) as new_nodes
FROM nodes
WHERE created_at >= datetime('now', '-8 weeks')
GROUP BY week
ORDER BY week
""").fetchall()]
# Firms with no person nodes (coverage gap)
metrics["firms_without_persons"] = conn.execute("""
SELECT COUNT(*) as count FROM nodes f
WHERE f.node_type = 'firm'
AND f.node_subtype IN ('public_pension', 'endowment', 'foundation')
AND NOT EXISTS (
SELECT 1 FROM edges e
WHERE e.to_node_id = f.id
AND e.edge_type IN ('TRUSTEE_OF', 'COMMITTEE_MEMBER_OF', 'WORKS_AT')
AND e.valid_to IS NULL
)
""").fetchone()[0]
# Pending entity resolution items
metrics["resolution_queue_pending"] = conn.execute("""
SELECT COUNT(*) FROM resolution_queue WHERE status = 'pending'
""").fetchone()[0]
# Stale data: nodes last updated >90 days ago that are high-priority
ninety_days_ago = (datetime.utcnow() - timedelta(days=90)).isoformat()
metrics["stale_high_priority_nodes"] = conn.execute("""
SELECT COUNT(*) FROM nodes
WHERE node_subtype IN ('public_pension', 'endowment')
AND updated_at < ?
""", (ninety_days_ago,)).fetchone()[0]
conn.close()
return metrics
def print_coverage_report():
report = harvest_coverage_report()
print("=" * 60)
print("INVESTOR GRAPH — COVERAGE REPORT")
print(f"Generated: {datetime.utcnow().isoformat()}")
print("=" * 60)
print("\nNode counts by type:")
for t, c in report["node_counts"].items():
print(f" {t:<20} {c:>8,}")
print("\nFirm nodes by subtype:")
for t, c in report.get("firm_subtypes", {}).items():
print(f" {t:<30} {c:>6,}")
print("\nActive edge counts:")
for t, c in report["active_edge_counts"].items():
print(f" {t:<30} {c:>6,}")
print("\nRecent harvest activity (30 days):")
for h in report["recent_harvests"]:
print(f" {h['source_type']:<20} {h['harvest_count']:>4} runs, "
f"{h['unique_urls']:>4} unique URLs")
print("\nNew nodes per week:")
for w in report["new_nodes_weekly"]:
bar = "█" * min(w["new_nodes"] // 5, 40)
print(f" {w['week']} {bar} ({w['new_nodes']})")
print(f"\nFirms without any person coverage: {report['firms_without_persons']}")
print(f"Entity resolution queue (pending): {report['resolution_queue_pending']}")
print(f"Stale high-priority nodes (>90d): {report['stale_high_priority_nodes']}")
if __name__ == "__main__":
print_coverage_report()
Integration Summary: Data Flow Diagram
[ browser-use Agent ]
|
| 1. Navigates target sites
| 2. Extracts structured JSON
| 3. Discovers PDF links
v
[ HarvestResult JSON ]
|
+----> [ ingest_harvest_result() ]
| |
| v
| [ EntityResolver ]
| |
| v
| [ SQLite Graph DB ] (nodes + edges + sources)
|
+----> [ download_pdf() ]
|
v
[ /raw/browser-harvest/pdfs/ ]
|
v
[ queue_for_chandra() ]
|
v
[ scripts/ingest_pdfs.py ] (Chandra OCR pipeline)
|
v
[ .chandra.txt sidecar ]
|
v
[ spaCy NER extraction ]
|
v
[ Person nodes / edges ]
|
v
[ SQLite Graph DB ]
|
v
[ Clay.com waterfall enrichment ]
(email, phone, LinkedIn verification)
|
v
[ Updated node metadata + confidence scores ]
Key Implementation Notes
Import resolution: The code examples above reference HarvestResult, HarvestOrganization, HarvestPerson, HarvestRelationship from a shared harvest_models.py file — consolidate those dataclasses there. Also import ingest_harvest_result, flag_for_review, get_browser_config, download_pdf from a shared harvest_utils.py.
Task prompt versioning: Task strings should be versioned in a tasks/ directory (e.g., tasks/calpers_board_v2.txt) and loaded at runtime. When site structure changes break extraction, increment the version and preserve the old task for audit. This prevents silent regressions where a task returns empty results without error.
Confidence score calibration: The default web scrape confidence of 0.70 is intentionally below the 0.85 threshold used for Form 5500 and 0.90 for EDGAR ADV data. When a node has both a web-scrape edge and a Form 5500 edge, the 5500 edge takes precedence in any conflict. The weight column on edges enables this precedence calculation at query time.
Clay enrichment integration: After each harvest batch, export new person nodes (created_at > last Clay export timestamp) as a CSV with name, title, organization, and source_url. Import into Clay.com for waterfall enrichment (email, direct phone, LinkedIn URL). Map Clay output back to node metadata updates and alias entries. This is the final layer that converts graph intelligence into contactable leads.
Sources and References
- browser-use GitHub repository — primary library documentation
- Playwright Python documentation — underlying automation framework
- browser-use documentation — task specification patterns, Controller API, BrowserConfig reference
- APScheduler documentation — scheduler configuration
- Litestream — SQLite WAL replication to S3 for durability
- SQLite Graph DB schema — canonical graph schema this pipeline feeds
- Institutional Investor Board Datasets — target site inventory and data quality notes
- Pillar PDF Ingestion Notes — Chandra OCR pipeline operating constraints (GPU serialization, cache behavior)
- CalPERS Board of Administration
- SEC EDGAR Full-Text Search API
- DOL Form 5500 Datasets
- ProPublica Nonprofit Explorer API
- NACUBO Annual Conference
- NCPERS Annual Conference
- P&I Investment Consulting Conference