ai-agent-vcr 0.1.1

The DVR for AI Agents - Record, visualize, and time-travel through agent execution

📼 Agent VCR

Time-travel debugging for AI agents.

📖 Documentation • 🚀 Examples

---

🛑 The Problem

Building multi-step AI agents (like LangGraph or CrewAI) is painfully slow.

When your agent fails on step 8 out of 10, traditional observability tools only tell you what went wrong. To fix it, you have to patch the prompt or code and re-run all 10 steps from the beginning.

Every typo or logic error costs you minutes of waiting and dollars in wasted LLM tokens.

💡 The Solution

Agent VCR makes debugging instant.

We record your agent's state at every step. When a failure happens, you simply rewind to the failing step, edit the state to fix the bug, and resume execution from that exact point.

LangSmith and LangFuse show you what happened. Agent VCR lets you change it.

• 🔌 Plug & Play: 1-line integration with LangGraph and others.
• 🚀 Zero Overhead: <5ms latency penalty per step.
• 📁 No Vendor Lock-in: Stores runs locally as git-friendly JSONL.
• 🔄 Async Native: Built from the ground up for modern asyncio agents.

---

🔥 Quick Start

pip install ai-agent-vcr

from agent_vcr import VCRRecorder, VCRPlayer

# 1. Record your agent (One-time setup)
recorder = VCRRecorder()
recorder.start_session("bug_hunt")
# ... your agent code runs here ...
recorder.save()

# 2. Time-Travel & Fix (The magic part)
player = VCRPlayer.load(".vcr/bug_hunt.vcr")

state = player.goto_frame(2)    # Jump back to step 2
state["prompt"] = "Fixed!"      # Fix the bad state
player.resume(from_frame=2)     # Resume execution from step 2

Features

• 🔴 Live Recording — Watch your agent execute in real-time via WebSocket
• ⏮️ Time Travel — Jump to any step, inspect full state
• ✏️ State Injection — Edit state and resume execution
• 🌳 DAG Visualization — See parallel execution branches
• 🔌 Framework Agnostic — Works with LangGraph, CrewAI, or raw Python
• 📁 Git-Friendly Format — JSONL files, version controllable
• 🚀 Production Performance — <5ms overhead per frame
• 🔄 Async-First — Full async recorder and player support

---

Who Is This For?

If you are...	Agent VCR helps you...
An AI engineer debugging LangGraph agents	Rewind to the exact failing step, fix state, and resume — no re-running the whole chain
A team lead reviewing agent behavior	Compare two execution paths side-by-side with full state diffs
A researcher iterating on prompts	Fork from any step, change the prompt, and see how downstream behavior changes
Building production agents	Record every execution in JSONL for audit trails and regression testing

---

How Does It Compare?

Feature	Agent VCR	LangSmith	LangFuse	Arize Phoenix
Record execution traces	✅	✅	✅	✅
Time-travel to any step	✅	❌	❌	❌
Edit state & resume	✅	❌	❌	❌
Fork from any frame	✅	❌	❌	❌
Compare execution runs	✅	✅	⚠️	⚠️
Self-hosted / local-first	✅	❌	✅	✅
Git-friendly format (JSONL)	✅	❌	❌	❌
Framework agnostic	✅	⚠️ LangChain	✅	✅
Zero external dependencies	✅	❌ Cloud	❌ Cloud	✅
Setup lines	3	~15	~10	~10

---

Framework Integrations

LangGraph

from langgraph.graph import StateGraph
from agent_vcr import VCRRecorder
from agent_vcr.integrations.langgraph import VCRLangGraph

# Your existing LangGraph code
graph = StateGraph()
graph.add_node("planner", planner_node)
graph.add_node("coder", coder_node)
graph.add_edge("planner", "coder")

# Add VCR recording with one line
recorder = VCRRecorder()
graph = VCRLangGraph(recorder).wrap_graph(graph)

# Run normally — recording happens automatically
result = graph.invoke({"query": "Build a todo app"})

Raw Python

from agent_vcr.integrations.langgraph import vcr_record

recorder = VCRRecorder()

@vcr_record(recorder, node_name="my_function")
def my_function(data):
    return process(data)

# Each call is automatically recorded
result = my_function({"key": "value"})

CrewAI

from crewai import Crew, Agent, Task
from agent_vcr import VCRRecorder
from agent_vcr.integrations.crewai import VCRCrewAI, vcr_task

recorder = VCRRecorder()
recorder.start_session("crew_debug_run")

# Option 1: Wrap the whole crew (auto-records every task)
crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
vcr_crew = VCRCrewAI(recorder)
result = vcr_crew.kickoff(crew)

recorder.save()

# Option 2: Decorate individual task functions
@vcr_task(recorder, task_name="research_step")
def research(context: dict) -> str:
    return "findings..."

Install with:

pip install ai-agent-vcr[crewai]

See examples/crewai_integration.py for a full runnable demo.

---

Storage Format

Agent VCR uses JSONL (JSON Lines) for storage:

{"type": "session", "data": {"session_id": "abc123", "created_at": "2024-01-01T00:00:00Z", ...}}
{"type": "frame", "data": {"frame_id": "...", "node_name": "planner", "input_state": {...}, "output_state": {...}, ...}}
{"type": "frame", "data": {...}}

Benefits:

• ✅ Human-readable
• ✅ Git-diffable
• ✅ Append-only (efficient for streaming)
• ✅ Line-by-line parsing (no need to load entire file)

---

Performance

Performance is continuously benchmarked in CI to ensure <5ms recording overhead.

To run the reproducible benchmarks on your own hardware:

pytest tests/benchmarks/ -v

---

API Reference

VCRRecorder

class VCRRecorder:
def start_session(
    self,
    session_id: str = None,
    parent_session_id: str = None,
    forked_from_frame: int = None,
    metadata: dict = None,
    tags: list[str] = None,
) -> Session

def record_step(
    self,
    node_name: str,
    input_state: dict,
    output_state: dict,
    metadata: FrameMetadata = None,
    frame_type: FrameType = FrameType.NODE_EXECUTION,
) -> Frame

def record_llm_call(...)
def record_tool_call(...)
def record_error(...)
def save(self) -> Path
def fork(self, from_frame: int, ...) -> VCRRecorder

VCRPlayer

class VCRPlayer:
@classmethod
def load(cls, filepath: str) -> VCRPlayer

def goto_frame(self, index: int) -> dict
def get_frame(self, index: int) -> Frame
def list_nodes(self) -> list[str]
def get_errors(self) -> list[Frame]
def compare_frames(self, a: int, b: int) -> dict
def resume(self, agent_callable: Callable, config: ResumeConfig) -> str
def export_state(self, frame_index: int) -> dict

ResumeConfig

class ResumeConfig:
from_frame: int              # Frame to resume from
new_session_id: str = None   # Optional ID for forked session
state_overrides: dict = {}   # State changes to apply
mode: ResumeMode = FORK      # FORK, REPLAY, or MOCK
skip_nodes: list[str] = []   # Nodes to skip during replay
inject_mocks: dict = {}      # Mock values for dependencies

---

Examples

See the examples/ directory for:

• basic_usage.py — Simple recording and playback
• time_travel_demo.py — Full time-travel workflow
• langgraph_integration.py — LangGraph auto-instrumentation

Run an example:

python examples/time_travel_demo.py

---

Contributing

We welcome contributions! Please see CONTRIBUTING.md for guidelines.

Development Setup

git clone https://github.com/agent-vcr/agent-vcr.git
cd agent-vcr
pip install -e ".[dev]"

Running Tests

# Unit tests
pytest tests/unit/ -v

# Integration tests
pytest tests/integration/ -v

# E2E tests
pytest tests/e2e/ -v

# Benchmarks
pytest tests/benchmarks/ -v

# With coverage
pytest --cov=agent_vcr --cov-report=html

---

Roadmap

• Core recording and playback
• Time-travel resume
• FastAPI server with WebSocket
• LangGraph integration
• Async recorder and player
• Terminal TUI debugger (vcr-tui)
• CI/CD integrations
• React dashboard
• CrewAI integration
• AutoGen integration
• Cloud storage backend
• Collaborative debugging

---

License

MIT License — see LICENSE for details.

---

Acknowledgments

Inspired by:

• LangSmith — For the observability paradigm
• GDB — For the time-travel debugging concept
• Chrome DevTools — For the UX patterns

---

Built with ❤️ by the Agent VCR community