GraphIngest SDK Documentation

# pipeline.py — save this file and run: python pipeline.py
from graphingest import node, graph, deploy, RetryPolicy
import requests

# ── Step 1: Define your nodes (individual tasks) ──

@node(name="fetch-page", cache_ttl=3600, max_retries=3)
def fetch_page(url: str) -> dict:
    """Fetch a web page. Cached for 1 hour. Retries 3x on failure."""
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return {"url": url, "status": resp.status_code, "length": len(resp.text)}

@node(name="summarize")
def summarize(page: dict) -> str:
    """Summarize a fetched page."""
    return f"Page {page['url']}: {page['length']} chars, status {page['status']}"

# ── Step 2: Compose nodes into a graph (pipeline) ──

@graph(
    name="web-scraper",
    retry_policy=RetryPolicy(
        max_retries=2,
        delay_seconds=1,
        backoff_factor=2,
        jitter=True,
    ),
    timeout_seconds=300,
)
def scrape_pipeline(urls: list[str]):
    # Fan-out: fetch all URLs in parallel
    pages = fetch_page.map(urls)

    # Process each result
    summaries = [summarize(page) for page in pages]

    return {"total": len(summaries), "summaries": summaries}

# ── Step 3: Deploy and run ──

deploy()  # push code to platform

# Execute the pipeline
result = scrape_pipeline([
    "https://example.com",
    "https://httpbin.org/get",
    "https://jsonplaceholder.typicode.com/posts/1",
])
print(result)
# → {"total": 3, "summaries": ["Page ...: 1256 chars, status 200", ...]}

pip install graphingest

# With AI agent support:
pip install graphingest[react]

# With LangGraph integration:
pip install graphingest[langgraph]

@node(name="extract-data", cache_ttl=3600, max_retries=5, tags=["etl"])
def extract(url: str) -> dict:
    response = requests.get(url)
    return response.json()

# Async support
@node(name="async-fetch")
async def fetch(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.json()

@graph(name="parallel-pipeline")
def pipeline(urls: list[str]):
    # Fan-out: dispatches len(urls) parallel workers
    results = extract.map(urls)
    # results[0] corresponds to urls[0], etc.
    return results

@graph(name="async-pipeline")
def pipeline(data: dict):
    future = slow_transform.submit(data)  # returns immediately
    do_other_stuff()                       # do work while it runs
    result = future.result(timeout=120)    # block until ready
    return result

RetryPolicy(
    max_retries=4,          # Total retry attempts
    delay_seconds=2,        # Initial delay
    backoff_factor=2.0,     # Multiplier per attempt
    max_delay_seconds=120,  # Upper bound
    jitter=True,            # ±50% randomization
)

# Delay formula:
# min(delay × backoff^attempt, max_delay) × random(0.5, 1.5)

@graph(name="sub-etl", retry_policy=RetryPolicy(max_retries=2))
def sub_etl(url: str) -> dict:
    data = extract(url)
    return transform(data)

@graph(name="main-pipeline")
def main_pipeline(urls: list[str]):
    results = []
    for url in urls:
        result = sub_etl(url)  # each creates a child graph run
        results.append(result)
    return results

@graph(name="my-pipeline", version="2.0", tags=["prod"])
def pipeline(source: str):
    ctx = GraphRunContext.get()

    ctx.graph_run_id         # "uuid-..."
    ctx.graph_name           # "my-pipeline"
    ctx.graph_version        # "2.0"
    ctx.parameters           # {"source": "..."}
    ctx.tags                 # ["prod"]
    ctx.parent_graph_run_id  # None (or parent's ID)

@node(name="extract")
def extract(url: str):
    ctx = NodeRunContext.get()

    ctx.node_run_id    # "uuid-..."
    ctx.node_key       # "extract"
    ctx.graph_run_id   # parent graph's run ID
    ctx.map_index      # None (or int if .map())
    ctx.retry_count    # 0

import logging
logger = logging.getLogger(__name__)

@node(name="extract")
def extract(url: str):
    logger.info("Starting extraction")   # → appears in dashboard
    logger.warning("Rate limit hit")
    logger.error("Connection failed")

# Or use the dedicated logger:
from graphingest import get_run_logger
log = get_run_logger()
log.info("Pipeline started")

from graphingest.langgraph import agent_node, agent_graph, AgentConfig

researcher = agent_node(
    name="researcher",
    graph_builder=build_research_agent,
    config=AgentConfig(
        model="gpt-4o",
        temperature=0.0,
        max_iterations=10,
        system_prompt="You are a research assistant.",
        stream_steps=True,
    ),
    cache_ttl=600,
    retries=2,
)

# Use like any node:
result = researcher("What is quantum computing?")

# Fan-out: run 50 agents in parallel
results = researcher.map(["query1", "query2", ..., "query50"])