Metadata-Version: 2.5
Name: ds-agentic-flow
Version: 0.38.0
Summary: Minimal wrapper over OpenAI Agent SDK with method chaining and ChatKit integration
Project-URL: Homepage, https://github.com/daiichisankyo/AgenticFlow
Project-URL: Documentation, https://daiichisankyo.github.io/AgenticFlow/
Project-URL: Repository, https://github.com/daiichisankyo/AgenticFlow
Project-URL: Issues, https://github.com/daiichisankyo/AgenticFlow/issues
Author-email: SnowLightPath <takuma.shibahara@daiichisankyo.com>
License: MIT
License-File: LICENSE
Keywords: agent,async,chatkit,llm,openai,workflow
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: openai-agents<0.15,>=0.14.0
Requires-Dist: openai-chatkit<2,>=1.5.3
Description-Content-Type: text/markdown

# AF - Agentic Flow Framework

[![Python 3.12+](https://img.shields.io/badge/python-3.12+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Docs](https://img.shields.io/badge/docs-online-green.svg)](https://daiichisankyo.github.io/AgenticFlow/)

> **🚀 v0.38 — SDK 0.14 Refresh.** AF now targets `openai-agents>=0.14.0,<0.15` (was `>=0.3.2`). Adds **`af.SandboxAgent`** for persistent containerized workspaces, drop-in support for the SDK's new session backends (`SQLAlchemy` / `MongoDB` / `Redis` / `Dapr` / `encrypted`), and the new `RunConfig` fields surfaced in 0.14. Existing `af.Agent` flows are unchanged. → [What's new](https://daiichisankyo.github.io/AgenticFlow/refresh-0.14/)
>
> Examples now default to **GPT-5.5** (released 2026-04-23). No code change required — just pass `model="gpt-5.5"`.

**Write agent workflows like regular Python code.**

```python
async def research_flow(query: str) -> str:
    # Internal thinking - not saved to session
    async with af.phase("Research"):
        findings = await researcher(query).stream()

    # persist=True saves the final response to session
    async with af.phase("Analysis", persist=True):
        return await analyst(findings).stream()
```

No graphs. No YAML. No state machines. Just Python.

---

## ✨ Why the agentic flow approach?

OpenAI Agents SDK is powerful, but multi-agent workflows get verbose fast:

<table>
<tr>
<th>Pure SDK (~50 lines)</th>
<th>AF (~15 lines)</th>
</tr>
<tr>
<td>

```python
# Manual orchestration
result1 = await Runner.run(
    researcher, messages
)
research = result1.final_output

result2 = await Runner.run(
    analyst,
    [{"role": "user", "content": research}]
)
# No streaming, no phases,
# no session management...
```

</td>
<td>

```python
async def my_flow(query: str) -> str:
    # Internal thinking - not saved to session
    async with af.phase("Research"):
        research = await researcher(query).stream()

    # persist=True saves the final response to session
    async with af.phase("Analysis", persist=True):
        return await analyst(research).stream()

runner = af.Runner(flow=my_flow)
result = await runner(query)
```

</td>
</tr>
</table>

**The agentic flow approach gives you:**

- **Callable agents** - `agent(prompt).stream()` like PyTorch modules
- **Workflow phases** - Structure and visibility for multi-step processes
- **Streaming mode** - `.stream()` for faster first-token via streaming API
- **Session injection** - Conversation persistence without global state
- **ChatKit integration** - Production-ready UI with full-text display

---

## 🆕 New: `.snapshot()` — Parallel Agents That Actually Know What's Going On

Ever tried running agents in parallel with `asyncio.gather()`? You probably hit this tradeoff:

| | Knows Context | Parallel-Safe |
|---|:---:|:---:|
| Default | ✅ | ❌ Race conditions! |
| `.isolated()` | ❌ Amnesia... | ✅ |
| **`.snapshot()`** | **✅** | **✅ Best of both!** |

`.snapshot()` gives each agent a **read-only snapshot** of the conversation — they see everything that happened before, but their responses don't interfere with each other.

```python
async with af.phase("Parallel Analysis", persist=True):
    # 3 agents, 1 context, 0 race conditions
    sentiment, entities, summary = await asyncio.gather(
        sentiment_agent(data).snapshot(),   # ← reads context
        entity_agent(data).snapshot(),      # ← reads context
        summary_agent(data).snapshot(),     # ← reads context
    )
    # All three see the same conversation history.
    # None of them write to it. No conflicts. Just works.
```

> **TL;DR:** `.isolated()` = no memory, `.snapshot()` = read-only memory, default = full read/write.
> Pick the right one for your use case. [Learn more →](https://daiichisankyo.github.io/AgenticFlow/concepts/modifiers/)

---

## 🆕 New: Multi-Provider Support

Use any LLM provider through the SDK's built-in multi-provider system:

```python
# OpenAI (default)
openai_agent = af.Agent(name="openai", model="gpt-5.5", instructions="...")

# Anthropic via LiteLLM
claude_agent = af.Agent(name="claude", model="litellm/anthropic/claude-sonnet-4-20250514", instructions="...")

# Runtime model switch
from agents import RunConfig
result = await agent("prompt").run_config(RunConfig(model="gpt-5.5")).stream()
```

No AF-specific configuration needed — the SDK handles provider routing via model name prefixes. See [Multi-Provider Guide](https://daiichisankyo.github.io/AgenticFlow/guides/multi-provider/).

> **Security:** LiteLLM 1.82.7/1.82.8 were compromised on PyPI (credential exfiltration). Install with `pip install "litellm>=1.82.6,!=1.82.7,!=1.82.8"` and audit before upgrading.

---

## 🆕 New: Sandbox Agents — Agents That Live in a Workspace

`agents.sandbox.SandboxAgent` (introduced in openai-agents 0.14) gives an agent a **persistent containerized workspace** — files survive across calls, state can be snapshotted and resumed. AF wraps it as `af.SandboxAgent`, which is an `af.Agent` subclass: every existing modifier (`.stream()`, `.silent()`, `.snapshot()`, `.isolated()`, `.max_turns()`, `.context()`, `.run_config()`) just works.

```python
import agentic_flow as af
from agents import RunConfig, SQLiteSession
from agents.sandbox import Manifest, SandboxRunConfig

# Plan agent: thinking only — no sandbox needed
planner = af.Agent(
    name="planner",
    instructions="Read the spec and produce a step-by-step implementation plan.",
    model="gpt-5.5",
)

# Coder agent: persistent workspace declared as part of agent identity
coder = af.SandboxAgent(
    name="coder",
    instructions="Implement the plan in Python.",
    model="gpt-5.5",
    default_manifest=Manifest(version="1", root="/work", entries=[]),
)

# Flow is just business logic — it does not know about execution environment
async def implement_flow(spec: str) -> str:
    async with af.phase("Plan"):
        plan = await planner(spec).stream()
    async with af.phase("Implement", persist=True):
        return await coder(plan).stream()

# Runner injects the execution environment (Session + sandbox transport) via contextvars
runner = af.Runner(
    flow=implement_flow,
    session=SQLiteSession("chat.db"),
    default_run_config=RunConfig(sandbox=SandboxRunConfig(...)),
)
```

`Manifest` is the workspace declaration and belongs to the agent (WHAT). `SandboxRunConfig` is the runtime transport (which sandbox client, which concurrency limits, which snapshot to resume) and belongs to the Runner (WHERE-environment). Per-call overrides remain available via the existing `.run_config()` modifier — Runner default + ExecutionSpec modifier compose so that the most-specific setting wins. `Manifest`, `SandboxRunConfig`, and the memory/snapshot configs come straight from the SDK; AF intentionally doesn't re-wrap them. [Learn more →](https://daiichisankyo.github.io/AgenticFlow/concepts/sandbox/)

---

## 📋 Requirements

- Python 3.12+
- [uv](https://docs.astral.sh/uv/) (recommended) or pip
- OpenAI API key

---

## 📦 Installation

### With uv (recommended)

```bash
# Install uv if you don't have it
curl -LsSf https://astral.sh/uv/install.sh | sh

# Add AF to your project
uv add git+https://github.com/daiichisankyo/AgenticFlow.git
```

### With pip (alternative)

```bash
pip install git+https://github.com/daiichisankyo/AgenticFlow.git
```

### Set your API key

Create a `.env.local` file from the example:

```bash
cp .env.example .env.local
# Edit .env.local and add your OpenAI API key
```

Or set it directly in your environment:

```bash
export OPENAI_API_KEY="your-api-key"
```

---

## 🚀 Quickstart Code

```python
import agentic_flow as af

# Define agents
researcher = af.Agent(
    name="researcher",
    instructions="Research the topic thoroughly.",
    model="gpt-5.5",
)

writer = af.Agent(
    name="writer",
    instructions="Write clear, engaging content.",
    model="gpt-5.5",
)

# Define workflow as a regular async function
async def blog_flow(topic: str) -> str:
    # Internal thinking - not saved to session
    async with af.phase("Research"):
        research = await researcher(topic).stream()

    # persist=True saves the final response to session
    async with af.phase("Writing", persist=True):
        return await writer(f"Write about: {research}").stream()

# Run
runner = af.Runner(flow=blog_flow)
article = await runner("quantum computing")
```

---

## 💡 Core Concepts

AF is built on three primitives:

- **Agent** - Callable wrapper around SDK Agent. Returns `ExecutionSpec` for deferred execution.
- **SandboxAgent** - Subclass of `Agent` that constructs `agents.sandbox.SandboxAgent` (persistent workspace, snapshots, sandbox memory). Uses the same `ExecutionSpec` and modifiers.
- **ExecutionSpec** - Lazy specification configured with modifiers (`.stream()`, `.isolated()`, `.snapshot()`, `.silent()`, `.max_turns()`)
- **phase** - Context manager for workflow boundaries. Controls session persistence with `persist=True`.

For details, see [Concepts](https://daiichisankyo.github.io/AgenticFlow/concepts/).

---

## 🔄 Patterns

### Sequential Pipeline

```python
async def pipeline(input: str) -> str:
    # Internal processing - not saved to session
    async with af.phase("Extract"):
        entities = await extractor(input).stream()

    async with af.phase("Enrich"):
        enriched = await enricher(entities).stream()

    # persist=True saves the final response to session
    async with af.phase("Format", persist=True):
        return await formatter(enriched).stream()
```

### Conditional Branching

```python
async def smart_process(data: str) -> str:
    # Internal classification - not saved to session
    async with af.phase("Classify"):
        category = await classifier(data).stream()

    # persist=True on whichever branch returns the final response
    if "technical" in category:
        async with af.phase("Technical Analysis", persist=True):
            return await tech_analyst(data).stream()
    else:
        async with af.phase("General Summary", persist=True):
            return await summarizer(data).stream()
```

### Iterative Refinement

```python
async def refine(draft: str) -> str:
    for i in range(3):
        # Internal review - not saved to session
        async with af.phase(f"Review #{i+1}"):
            feedback = await critic(draft).stream()

        if "APPROVED" in feedback:
            return draft

        # Internal revision - not saved to session
        async with af.phase(f"Revise #{i+1}"):
            draft = await reviser(f"{draft}\n\nFeedback: {feedback}").stream()

    # Note: This pattern returns the draft directly without session save.
    # Add a final phase with persist=True if you need to save the result.
    return draft
```

### Parallel Execution

```python
import asyncio

async def parallel_analysis(data: str) -> dict:
    async with af.phase("Parallel Analysis", persist=True):
        # snapshot() reads phase context but doesn't write (concurrent-safe)
        sentiment, entities, summary = await asyncio.gather(
            sentiment_agent(data).snapshot(),
            entity_agent(data).snapshot(),
            summary_agent(data).snapshot(),
        )

        return await synthesizer(
            f"Sentiment: {sentiment}\nEntities: {entities}\nSummary: {summary}"
        ).stream()
```

---

## 🔌 ChatKit Integration

```python
from pathlib import Path
import agentic_flow as af
from agents import SQLiteSession
from chatkit.server import ChatKitServer
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

# Persist data in a known location
DATA_DIR = Path(__file__).parent / "data"
DATA_DIR.mkdir(exist_ok=True)

app = FastAPI()


class MyServer(ChatKitServer):
    async def respond(self, thread, item, context):
        user_message = item.content[0].text if item else ""
        session = SQLiteSession(
            session_id=thread.id,
            db_path=str(DATA_DIR / "chat.db"),
        )
        runner = af.Runner(flow=my_flow, session=session)
        async for event in af.chatkit.run_with_chatkit_context(
            runner, thread, self.store, context, user_message
        ):
            yield event


server = MyServer(store)


@app.post("/chatkit")
async def chatkit_endpoint(request: Request):
    result = await server.process(await request.body(), {})
    return StreamingResponse(result, media_type="text/event-stream")
```

Each `af.phase()` automatically creates workflow boundaries for proper reasoning display.

---

## 🎮 Try the Demos

First, install sample dependencies:

```bash
uv sync --group sample
```

### CLI Quickstart

Experience the core concepts interactively:

```bash
uv run --group sample python sample/quickstart.py
```

Demonstrates:
1. Flow & Runner separation
2. Declaration vs execution (`agent(prompt)` returns spec, `await` executes)
3. Modifiers (`.stream()`, `.isolated()`, `.silent()`)
4. Typed output with Pydantic

### Guide TUI

Interactive Textual UI that answers questions about AF:

```bash
uv run --group sample python -m sample.guide.cli
```

### Guide Web Server (FastAPI + ChatKit)

Start the backend API:

```bash
uv run --group sample uvicorn sample.guide.server:app --reload --port 8000
```

### Guide Frontend (Next.js + ChatKit)

In a separate terminal:

```bash
cd sample/guide/frontend
npm install
npm run dev
```

Visit http://localhost:3000

---

## 📚 Documentation

- **[Concepts](https://daiichisankyo.github.io/AgenticFlow/concepts/)** - Call-Spec Discipline, ExecutionSpec, Phase, Modifiers
- **[API Reference](https://daiichisankyo.github.io/AgenticFlow/reference/api/)** - Complete API documentation
- **[Context Resolution](https://daiichisankyo.github.io/AgenticFlow/context-resolution/)** - Advanced context management

For the complete documentation site, visit [https://daiichisankyo.github.io/AgenticFlow/](https://daiichisankyo.github.io/AgenticFlow/)

---

## Contributing

```bash
# Clone and setup
git clone https://github.com/daiichisankyo/AgenticFlow.git
cd AgenticFlow
uv sync --group dev

# Run tests
uv run pytest

# Run linter
uv run ruff check src/
```

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Run tests and linter
5. Submit a pull request

---

## 📄 License

MIT - see [LICENSE](LICENSE) for details.
