# DataFlow-Agent
**Repository Path**: yonja/DataFlow-Agent
## Basic Information
- **Project Name**: DataFlow-Agent
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-28
- **Last Updated**: 2026-07-28
## Categories & Tags
**Categories**: Uncategorized
**Tags**: None
## README
# ๐บ DataFlow-Agent
**Synthesize, score, and refine agent trajectories โ high-quality training data for tool-using LLMs.**
Agentic-exploration operators for [DataFlow](https://github.com/OpenDCAI/DataFlow): drive an LLM agent through a **pluggable sandbox** to produce the
`task โ [thought โ tool call โ observation]* โ answer` data used to train and evaluate agentic models.
[](test/test_agentic_explore.py)
[](pyproject.toml)
[](https://github.com/OpenDCAI/DataFlow)
[](#)
[Quickstart](#-quickstart) ยท [Operators](#-the-pipeline) ยท [Sandboxes](#-pluggable-sandboxes) ยท [Docs](#-documentation) ยท [Roadmap](#-roadmap)
---
## โจ Why DataFlow-Agent
Most sandboxes only **collect** trajectories. DataFlow-Agent **collects, scores, selects, and repairs** them โ turning raw agent rollouts into training-grade data through a full pipeline:
```
seed tasks
โ
โโโโโโโโโโโโโโโโโโโผโโโโโโโโโโโโโโโโโโ
โ GENERATE explore a sandbox โ linear chain โโโบ AgentExploreGenerator
โ (think โ act โ observe)โ branching tree โโโบ AgentExploreTreeGenerator
โโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโ
โ trajectories
โโโโโโโโโโโโโโโโโโโผโโโโโโโโโโโโโโโโโโ
โ EVALUATE LLM-as-judge (4 axes) โ โโโบ TrajectoryQualityEvaluator
โโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโ
โ + quality scores
โโโโโโโโโโโโโโโโโโโผโโโโโโโโโโโโโโโโโโ
โ SELECT/FILTER keep the good ones โ โโโบ TrajectorySelector (top-N diverse)
โ โ TrajectoryFilter (rule-based gate)
โโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโ
โ high-quality set
โโโโโโโโโโโโโโโโโโโผโโโโโโโโโโโโโโโโโโ
โ REFINE repair the salvageable โ โโโบ TrajectoryRefiner
โโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโ
โผ
training-grade dataset
```
**Zero coupling.** Operators depend only on two abstractions โ `LLMServingABC` (decides / judges) and `SandboxClientABC` (executes). They never import a concrete sandbox. Swap the backend, the operators don't change.
---
## ๐ Quickstart
```bash
# 1. install (open-dataflow provides OperatorABC / LLMServingABC / registry / storage)
pip install open-dataflow
pip install -e .
# 2. run the offline test suite โ 42 tests, no network / GPU / API key
pytest test/test_agentic_explore.py -v
# 3. run the offline demo (mock sandbox + scripted LLM)
python examples/agentic_explore/run_mock_pipeline.py
```
`import dataflow_agent` registers every operator into DataFlow's `OPERATOR_REGISTRY`, so they resolve by name like any built-in operator.
### Wiring a real run
```python
import dataflow_agent # registers the operators
from dataflow.serving import APILLMServing_request # from open-dataflow
from dataflow.utils.storage import FileStorage
from dataflow_agent import AgentExploreGenerator, HTTPSandboxClient
storage = FileStorage(first_entry_file_name="queries.jsonl", cache_path="./cache")
llm = APILLMServing_request(api_url="https://.../v1/chat/completions", model_name="gpt-4o")
sandbox = HTTPSandboxClient(base_url="http://127.0.0.1:18890", domain="text2sql")
op = AgentExploreGenerator(llm_serving=llm, sandbox=sandbox, domain="text2sql",
max_steps=10, max_workers=8)
op.run(storage.step(), input_key="query", output_key="trajectory")
```
---
## ๐งฉ The pipeline
| Stage | Operator | What it does | LLM? |
|---|---|---|:--:|
| **Generate** | `AgentExploreGenerator` | Linear trajectory โ one thinkโactโobserve chain per task. | โ
|
| **Generate** | `AgentExploreTreeGenerator` | Branching **trajectory tree** โ samples N candidate actions per node, dedups, expands (depth/breadth/node-bounded, with a `depth_threshold`). Emits the tree **and** its root-to-leaf `paths` as linear trajectories. | โ
|
| **Evaluate** | `TrajectoryQualityEvaluator` | **LLM-as-judge** on 4 axes (goal / efficiency / coherence / tool-use, 1โ5) + `overall` โ [0,1] + rationale. | โ
|
| **Select** | `TrajectorySelector` | **Top-N diverse selection**: score by depth(40)+info(30)+tool-diversity(30), then Jaccard de-dup. Deterministic. | โ |
| **Filter** | `TrajectoryFilter` | **Rule-based quality gate** (success / step bounds / parse-error / hallucinated-tool / tool-error / repeated-action loop / empty answer). Deterministic. | โ |
| **Refine** | `TrajectoryRefiner` | Re-explores **failed / low-scoring** trajectories primed with a diagnosis of what went wrong; good ones pass through untouched. | โ
|
> **Select vs Filter** โ Filter judges each trajectory good/bad and drops the bad. Selector picks the best *N distinct* from a pool ("one seed โ one tree โ N gems").
### Output schema
Every trajectory row is a flat, judge-ready record:
```json
{
"task": "...",
"steps": [{"thought": "...", "action": {"tool": "...", "args": {...}}, "observation": ...}],
"final_answer": "...",
"num_steps": 3,
"success": true
}
```
---
## ๐ Pluggable sandboxes
A sandbox is any `SandboxClientABC` subclass. Adding one = implement `list_tools` + `execute`; the six operators stay untouched.
| Backend | Module | Use case |
|---|---|---|
| `CodingSandboxClient` | `sandbox/coding_client.py` | **Real coding agent** โ isolated workspace with `read_file` / `write_file` / `run_python` / `run_tests` (pytest) / `run_shell`. Fix bugs, implement functions, run tests. |
| `HTTPSandboxClient` | `sandbox/http_client.py` | Drives a remote sandbox server **over HTTP only** (plain `requests`) โ imports nothing from any sandbox SDK. Works with any server speaking the generic `{code,message,data,meta}` protocol (web / rag / text2sql / doc / ds domains). |
| `MockSandboxClient` | `sandbox/mock_client.py` | Offline, network-free. For tests / dev. |
| *your own* | add a subclass | Wrap any API / MCP server / tool as `ToolResult`. |
### Scope
A **text / structured-domain** explorer (web ยท rag ยท sql ยท doc ยท ds ยท coding ยท shell). Observations are fed back as text, so image/binary observations (GUI/VM `screenshot`) are **out of scope** โ that's the multimodal explorer on the roadmap. The transport layer itself is domain-agnostic.
---
## ๐ Documentation
| Doc | Contents |
|---|---|
| [`docs/DESIGN_zh.md`](docs/DESIGN_zh.md) | ่ฎพ่ฎก็ๅฟต โ ไธบไปไน่ฟไน่ฎพ่ฎกใไบๆฎตๆตๆฐด็บฟใๅฏๆๆๆฒ็ฎฑ |
| [`docs/CAPABILITIES_zh.md`](docs/CAPABILITIES_zh.md) | ่ฝๅๆธ
ๅ โ ๆฏๆๅชไบ็ฏๅขใ่ฝ็ๆๅชไบๆฐๆฎใๆ็ๅบฆ |
| [`examples/agentic_explore/`](examples/agentic_explore/) | Runnable examples โ mock pipeline, coding agent, filter demo, real-API e2e |
---
## ๐บ๏ธ Roadmap
- [x] Generator โ Evaluator โ Filter โ **Refiner** loop (repair, not just drop)
- [x] **TrajectorySelector** โ top-N diverse selection algorithm
- [x] **CodingSandboxClient** โ real workspace + pytest
- [ ] **Multimodal explorer** for GUI/VM (image observations)
- [ ] **Preference-pair export** (best vs. worst sibling paths โ DPO data)
---