Skip to content

Getting started

Install

pip install deepharness

Or from a clone, for local development:

git clone https://github.com/sayedshaun/deepharness.git
cd deepharness
pip install -e .

Requires Python 3.11+. The only runtime dependency is httpx.

Quickstart

Two lookups run at the same time, then a third node joins their results:

import asyncio
from dataclasses import dataclass

from deepharness import Graph


@dataclass
class State:
    sales: str = ""
    churn: str = ""
    summary: str = ""


graph = Graph(State)


@graph.add(start=True)
async def fetch_sales(state: State) -> State:
    await asyncio.sleep(0.5)  # a slow API call, an agent, whatever
    state.sales = "Sales up 12% QoQ."
    return state


@graph.add(start=True)
async def fetch_churn(state: State) -> State:
    await asyncio.sleep(0.3)
    state.churn = "Churn down to 4%."
    return state


@graph.add(end=True)
def summarize(state: State) -> State:
    state.summary = f"{state.sales} {state.churn}"
    return state


graph.connect(fetch_sales, summarize)
graph.connect(fetch_churn, summarize)

executor = graph.build()
result = asyncio.run(executor.run())

print(result.summary)  # Sales up 12% QoQ. Churn down to 4%.

Both fetch_* nodes are marked start=True, so they fire concurrently. summarize has two incoming edges, so it waits for both before running — a join, expressed by wiring alone.

Adding an agent

Drop an Agent inside any node and the graph handles sequencing around it:

from deepharness import Agent, Message
from deepharness import OpenAI

writer = Agent(OpenAI("gpt-4o-mini"), name="writer", system="You are concise.")


@graph.add(start=True, end=True)
async def write(state: State) -> State:
    out = await writer.arun(f"Summarize: {state.sales}")
    state.summary = out.output
    return state

Next steps