Weave

Getting Started

Install

The core has zero runtime dependencies. Provider SDKs are optional extras.

# Everything (dev tooling + all providers) in a virtualenv:
make install
source .venv/bin/activate

# Or just the core + the provider you need:
pip install "weaveflow[anthropic]"     # or [openai] / [google] / [mistral] / [ollama] / [deepseek] / [all]

Requires Python 3.10+.

Your first agent

An agent is an async function decorated with @agent. It declares typed input/output ports, optional capability tags, and an optional LLM "brain".

import asyncio
from weave import agent, DataType, Payload

@agent(
    name="shout",
    input=DataType.TEXT,
    output=DataType.TEXT,
    tags=["text", "transform"],
)
async def shout(ctx):
    return ctx.input.value.upper()

result = asyncio.run(shout.run(Payload.text("hello")))
print(result.value)   # "HELLO"

ctx is the AgentContext: it carries the input payload plus the injected brain, memory, and logger.

Add an LLM brain

Pass llm="provider:model". Then ctx.complete(...) is one line. Set the matching API key env var (e.g. ANTHROPIC_API_KEY).

@agent(
    name="summarizer",
    input=DataType.TEXT,
    output=DataType.TEXT,
    tags=["summarization"],
    llm="anthropic:claude-opus-4-8",
)
async def summarizer(ctx):
    return await ctx.complete(f"Summarize in one sentence:\n{ctx.input.value}")

Compose a pipeline

Chain agents left to right. Weave validates that each link is compatible before running anything, and auto-transforms compatible-but-different types between hops.

from weave import Pipeline

pipe = Pipeline([cleaner, summarizer])         # links validated up front
out = await pipe.run("a long document ...")    # returns a Payload
print(out.value)

Inspect a run with the local runner

LocalRunner runs a chain in-process and records a trace of every hop. There's no platform and no network beyond your configured LLM.

from weave import LocalRunner

trace = await LocalRunner().simulate([cleaner, summarizer], "a long document ...")
for hop in trace.hops:
    print(hop.agent, "->", hop.output.value, f"({hop.elapsed_ms:.1f} ms)")
print("final:", trace.output.value)

Try the offline example

make example       # runs examples/quickstart.py, no API key needed

Next