Metadata-Version: 2.5
Name: zonix
Version: 0.5.0
Summary: Explicit, serializable AI workflow primitives inspired by pydantic-ai.
Project-URL: Homepage, https://github.com/zongxi1115/zonix
Project-URL: Repository, https://github.com/zongxi1115/zonix
Project-URL: Issues, https://github.com/zongxi1115/zonix/issues
Author-email: zongxi1115 <zxwang1234321@gmail.com>
License: MIT
License-File: LICENSE
Keywords: agent,ai,llm,pydantic,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: jsonschema<5,>=4.23
Requires-Dist: pydantic<3,>=2.8
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.45; extra == 'anthropic'
Provides-Extra: dev
Requires-Dist: mypy>=1.13; extra == 'dev'
Requires-Dist: ruff>=0.8; extra == 'dev'
Provides-Extra: gemini
Requires-Dist: google-genai>=1.0; extra == 'gemini'
Provides-Extra: openai
Requires-Dist: openai>=1.68; extra == 'openai'
Provides-Extra: viz
Requires-Dist: graphviz>=0.20; extra == 'viz'
Description-Content-Type: text/markdown

# Zonix

![Zonix logo](https://raw.githubusercontent.com/zongxi1115/zonix/main/logo.png)

<p align="center"><strong>Call simply. Chain deeply. Trace everything.</strong></p>

Zonix is a Python agent and workflow framework. It borrows the clarity of
pydantic-ai's `Agent`, then adds `workflow`, `team`, and `router` primitives on
top of one shared execution model. Agents, workflows, and teams are all nodes
with the same `__call__` / `run` / `stream` surface, and they share trace, usage,
messages, and approval state.

The design goal is that a beginner can call one object and get a useful answer,
while an advanced user can turn on reasoning, usage accounting, raw provider
payloads, graph export, human approval, and frontend streaming without changing
the shape of their business code.

## Features

- **One execution model.** `await node(task)` returns the output, `node.run(task)`
  returns the full `RunResult`, `node.stream(task)` yields typed events. Agents,
  workflows, and teams all support the same three.
- **Typed structured output.** Set `output=SomeModel` and get a validated
  instance back, with automatic repair rounds when the model returns bad JSON.
- **Tools from type hints.** Schemas are generated from signatures and
  docstrings. Optional `ToolContext` injection, parallel execution, error
  capture, and middleware interception.
- **Human approval built in.** Mark a tool `approval=True` and `run()` returns a
  paused result you can `resume()`, or register an approver callback and keep the
  run going.
- **Approval while streaming.** Pass `approval=` to `stream()` and a pause is
  resolved by your handler mid-flight — the event stream keeps flowing instead of
  being cancelled.
- **Composable orchestration.** `workflow` gives `then`/`parallel`/`join`/
  `branch`/`loop`, plus plain functions as steps via `map`; `team` gives
  router-driven dispatch. Both export Mermaid, DOT, SVG, PNG, or PDF graphs, and
  both can share message context across members.
- **Resumable workflows.** `workflow(...).checkpoint(store)` persists each step's
  result; rerun with the same `run_id` and finished steps are skipped.
  `FileCheckpointStore` and `MemoryCheckpointStore` ship in the box, or implement
  the `Checkpointer` protocol yourself.
- **Explicit cancellation.** Every entry point takes `cancel=` (an
  `asyncio.Event`, or anything with `is_set()`), checked between workflow steps,
  team turns, model calls, and tool calls.
- **Provider-neutral adapters.** OpenAI (Chat and Responses), Anthropic, and
  Gemini, plus offline `Echo`/`StaticModel`/`ScriptedModel` for tests. Any
  OpenAI-compatible endpoint works by setting `base_url`.
- **Inspectable by default.** `RunResult` keeps the span tree, usage, messages,
  and the raw upstream request and response for every model call.
- **Sync facade.** `call_sync`, `run_sync`, `stream_sync` for scripts, CLIs, and
  notebooks — including resumable approvals.

## Install

```bash
pip install zonix
```

Optional provider extras:

```bash
pip install "zonix[openai]"
pip install "zonix[anthropic]"
pip install "zonix[gemini]"
pip install "zonix[viz]"     # image export for workflow/team graphs
```

Requires Python 3.11+. For local development from this repository: `pip install -e .`

## 60 seconds

```python
import asyncio
import os

from pydantic import BaseModel

from zonix import agent
from zonix.models import OpenAI


class Plan(BaseModel):
    goal: str
    files: list[str]
    steps: list[str]


planner = agent(
    "planner",
    role="Plan code work",
    model=OpenAI("gpt-5.5", api_key=os.environ["OPENAI_API_KEY"]),
    output=Plan,
)


@planner.tool
def read_tree(path: str) -> list[str]:
    """List files under a repository path."""
    return sorted(os.listdir(path))


async def main() -> None:
    plan = await planner("add captcha to the login page")
    print(plan.goal, plan.files)

    result = await planner.run("add captcha to the login page")
    print(result.usage.total_tokens)
    print(result.model_calls[-1].raw_response)

    async for event in planner.stream("add captcha to the login page"):
        print(event)


asyncio.run(main())
```

Any OpenAI-compatible endpoint works with the same adapter — only the model name
and `base_url` change:

```python
from zonix.models import OpenAI

deepseek = OpenAI(
    model="deepseek-chat",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com/v1",
)
```

Streaming chat requests automatically send `stream_options={"include_usage": true}`,
so token counts are complete even on streamed runs.

## Multi-agent

```python
from zonix import router, team, workflow
from zonix.types import Route

# Fixed pipeline: output of one step feeds the next.
flow = (
    workflow("review")
    .start(planner)
    .parallel(security_review, perf_review)
    .join(merge_reviews)
    .branch(lambda r: r.risk == "high", then=human_gate, else_=auto_apply)
    .build()
)

review = await flow("audit the auth changes", ctx=ctx)
print(flow.to_mermaid())


# Router-driven dispatch: the router picks the next node each step.
def choose(task, state) -> Route:
    if isinstance(task, Review):
        return Route(done=True)
    return Route(next="reviewer" if "review" in str(task).lower() else "coder")


code_team = (
    team("code_team")
    .add(planner, coder, reviewer)
    .route(router("rule_router", choose))
    .build(max_steps=6)
)

answer = await code_team("review the auth changes", ctx=ctx)
```

A router can be a rule function, another agent, or any node that returns
`Route(next=..., done=..., input=...)`. Workflows and teams are nodes themselves,
so they nest freely.

Long pipelines can checkpoint each step and resume where they stopped:

```python
from zonix import FileCheckpointStore

flow = (
    workflow("review")
    .start(planner)
    .map(lambda plan: plan.model_dump())    # plain functions are steps too
    .then(coder)
    .checkpoint(FileCheckpointStore("./.checkpoints"))
    .share_context()                        # each step sees the ones before it
    .build()
)

await flow("audit the auth changes", run_id="job-42")   # rerun skips finished steps
```

Checkpointed values round-trip as JSON, so a step that returned a `BaseModel`
replays as a `dict` — keep plain jsonable values flowing between steps, or
re-validate at the start of the next one.

## Human approval

```python
result = await coder.run("edit the login page", ctx=ctx)

if result.paused:
    print(result.pending.tool, result.pending.input)
    result = await result.resume(approve=True)
```

Or skip the pause entirely by passing an approver:

```python
result = await coder.run("edit the login page", approval=lambda pending: True)
```

Paused results hold a live continuation. Release them with `await result.cancel()`
(or `result.close()` for `run_sync`) if you will not resume.

Streaming takes the same handler, and the stream survives the pause:

```python
async for event in coder.stream("edit the login page", approval=my_handler):
    ...
```

Without a handler a streamed run emits `ApprovalRequired` and then cancels.

## Cancellation

```python
cancel = asyncio.Event()
task = asyncio.create_task(flow.run("long job", cancel=cancel))
cancel.set()     # cooperative: takes effect at the next checkpoint
```

## Tracing

Zonix ships vendor-neutral tracing hooks with no collector dependency. Register a
`SpanProcessor` once, then override per run:

```python
from zonix import TraceOptions, configure_tracing

configure_tracing(my_processor, defaults=TraceOptions(enabled=True, project="my-app"))

result = await planner.run(task, trace=TraceOptions(tags=["dev"], metadata={"user_id": "u1"}))
async for event in planner.stream(task, trace=TraceOptions(tags=["ui"])):
    ...
```

The span tree covers workflows, teams, agents, routers, model calls, and tool
calls. `result.trace` stays available even when export is disabled. The separate
`zonix-observe` package provides a local collector, storage, and browser UI.

## Architecture

```text
zonix/
  spec.py       agent()/team()/workflow()/router() factories
  engine.py     agent model and tool execution loop
  runtime.py    __call__/run/stream driver shared by every node
  types.py      Message, Usage, Span, RunState, RunResult, Route
  tools.py      tool definitions, ToolContext, middleware results
  graph.py      graph specs, Mermaid, DOT, and image export
  checkpoint.py Checkpointer protocol, file and in-memory stores
  memory/       Window, Summarize, Vector, Session
  multi/        Workflow, Team, Router nodes
  models/       OpenAI, Anthropic, Gemini, offline adapters
  hitl.py       approval keys and snapshot persistence
  tracing.py    vendor-neutral span lifecycle and processor hooks
  wire/         event-to-wire protocol adapters (Vercel AI SDK)
```

## Documentation

- [中文完整 API 教程](https://github.com/zongxi1115/zonix/blob/main/docs/tutorial.zh-CN.md)
- [Single agent example](https://github.com/zongxi1115/zonix/blob/main/examples/single_agent.py)
- [Workflow and team example](https://github.com/zongxi1115/zonix/blob/main/examples/workflow_team.py)
- [Real provider smoke script](https://github.com/zongxi1115/zonix/blob/main/scripts/smoke_real_provider.py)

## License

MIT
