vv-agent 0.1.0

Vector Vein inspired agent framework with cycle runtime, tools and memory management

v-agent

中文文档

A lightweight agent framework extracted from VectorVein's production runtime. Cycle-based execution with pluggable LLM backends, tool dispatch, memory compression, and distributed scheduling.

Architecture

AgentRuntime
├── CycleRunner          # single LLM turn: context -> completion -> tool calls
├── ToolCallRunner       # tool dispatch, directive convergence (finish/wait_user/continue)
├── RuntimeHookManager   # before/after hooks for LLM, tool calls, memory compaction
├── MemoryManager        # automatic history compression when context exceeds threshold
└── ExecutionBackend     # cycle loop scheduling
    ├── InlineBackend    # synchronous (default)
    ├── ThreadBackend    # thread pool with futures
    └── CeleryBackend    # distributed, per-cycle Celery task dispatch

Core types live in v_agent.types: AgentTask, AgentResult, Message, CycleRecord, ToolCall.

Task completion is tool-driven: the agent calls _task_finish or _ask_user to signal terminal states. No implicit "last message = answer" heuristics.

Setup

cp local_settings.example.py local_settings.py
# Fill in your API keys and endpoints in local_settings.py

uv sync --dev
uv run pytest

Quick Start

CLI

uv run v-agent --prompt "Summarize this framework" --backend moonshot --model kimi-k2.5

# With per-cycle logging
uv run v-agent --prompt "Summarize this framework" --backend moonshot --model kimi-k2.5 --verbose

CLI flags: --settings-file, --backend, --model, --verbose.

Programmatic

from v_agent.config import build_openai_llm_from_local_settings
from v_agent.runtime import AgentRuntime
from v_agent.tools import build_default_registry
from v_agent.types import AgentTask

llm, resolved = build_openai_llm_from_local_settings("local_settings.py", backend="moonshot", model="kimi-k2.5")
runtime = AgentRuntime(llm_client=llm, tool_registry=build_default_registry())

result = runtime.run(AgentTask(
    task_id="demo",
    model=resolved.model_id,
    system_prompt="You are a helpful assistant.",
    user_prompt="What is 1+1?",
))
print(result.status, result.final_answer)

SDK

from v_agent.sdk import AgentSDKClient, AgentSDKOptions

client = AgentSDKClient(options=AgentSDKOptions(
    settings_file="local_settings.py",
    default_backend="moonshot",
    default_model="kimi-k2.5",
))
result = client.run("Explain Python's GIL in one sentence.")
print(result.final_answer)

Execution Backends

The cycle loop is delegated to a pluggable ExecutionBackend.

Backend	Use case
InlineBackend	Default. Synchronous, single-process.
ThreadBackend	Thread pool. Non-blocking submit() returns a Future.
CeleryBackend	Distributed. Each cycle dispatched as an independent Celery task.

CeleryBackend

Two modes:

• Inline fallback (no RuntimeRecipe): cycles run in-process, same as InlineBackend.
• Distributed (with RuntimeRecipe): each cycle is a Celery task. Workers rebuild the AgentRuntime from the recipe and load state from a shared StateStore (SQLite or Redis).

from v_agent.runtime.backends.celery import CeleryBackend, RuntimeRecipe, register_cycle_task

register_cycle_task(celery_app)

recipe = RuntimeRecipe(
    settings_file="local_settings.py",
    backend="moonshot",
    model="kimi-k2.5",
    workspace="./workspace",
)
backend = CeleryBackend(celery_app=app, state_store=store, runtime_recipe=recipe)
runtime = AgentRuntime(llm_client=llm, tool_registry=registry, execution_backend=backend)

Install celery extras: uv sync --extra celery.

Cancellation and Streaming

from v_agent.runtime import CancellationToken, ExecutionContext

# Cancel from another thread
token = CancellationToken()
ctx = ExecutionContext(cancellation_token=token)
result = runtime.run(task, ctx=ctx)

# Stream LLM output token by token
ctx = ExecutionContext(stream_callback=lambda text: print(text, end=""))
result = runtime.run(task, ctx=ctx)

Workspace Backends

Workspace file I/O is delegated to a pluggable WorkspaceBackend protocol. All built-in file tools (_read_file, _write_file, _list_files, etc.) go through this abstraction.

Backend	Use case
LocalWorkspaceBackend	Default. Reads/writes to a local directory with path-escape protection.
MemoryWorkspaceBackend	Pure in-memory dict storage. Great for testing and sandboxed runs.
S3WorkspaceBackend	S3-compatible object storage (AWS S3, Aliyun OSS, MinIO, Cloudflare R2).

from v_agent.workspace import LocalWorkspaceBackend, MemoryWorkspaceBackend

# Explicit local backend
runtime = AgentRuntime(
    llm_client=llm,
    tool_registry=registry,
    workspace_backend=LocalWorkspaceBackend(Path("./workspace")),
)

# In-memory backend for testing
runtime = AgentRuntime(
    llm_client=llm,
    tool_registry=registry,
    workspace_backend=MemoryWorkspaceBackend(),
)

S3WorkspaceBackend

Install the optional S3 dependency: uv pip install 'v-agent[s3]'.

from v_agent.workspace import S3WorkspaceBackend

backend = S3WorkspaceBackend(
    bucket="my-bucket",
    prefix="agent-workspace",
    endpoint_url="https://oss-cn-hangzhou.aliyuncs.com",  # or None for AWS
    aws_access_key_id="...",
    aws_secret_access_key="...",
    addressing_style="virtual",  # "path" for MinIO
)

Custom Backend

Implement the WorkspaceBackend protocol (8 methods) to plug in any storage:

from v_agent.workspace import WorkspaceBackend

class MyBackend:
    def list_files(self, base: str, glob: str) -> list[str]: ...
    def read_text(self, path: str) -> str: ...
    def read_bytes(self, path: str) -> bytes: ...
    def write_text(self, path: str, content: str, *, append: bool = False) -> int: ...
    def file_info(self, path: str) -> FileInfo | None: ...
    def exists(self, path: str) -> bool: ...
    def is_file(self, path: str) -> bool: ...
    def mkdir(self, path: str) -> None: ...

Modules

Module	Description
v_agent.runtime.AgentRuntime	Top-level state machine (completed / wait_user / max_cycles / failed)
v_agent.runtime.CycleRunner	Single LLM turn and cycle record construction
v_agent.runtime.ToolCallRunner	Tool execution with directive convergence
v_agent.runtime.RuntimeHookManager	Hook dispatch (before/after LLM, tool call, memory compact)
v_agent.runtime.StateStore	Checkpoint persistence protocol (InMemoryStateStore / SqliteStateStore / RedisStateStore)
v_agent.memory.MemoryManager	Context compression when history exceeds threshold
v_agent.workspace	Pluggable file storage: LocalWorkspaceBackend, MemoryWorkspaceBackend, S3WorkspaceBackend
v_agent.tools	Built-in tools: workspace I/O, todo, bash, image, sub-agents, skills
v_agent.sdk	High-level SDK: AgentSDKClient, AgentSession, AgentResourceLoader
v_agent.skills	Agent Skills support (SKILL.md parsing, prompt injection, activation)
v_agent.llm.VVLlmClient	Unified LLM interface via vv-llm (endpoint rotation, retry, streaming)
v_agent.config	Model/endpoint/key resolution from local_settings.py

Built-in Tools

_list_files, _file_info, _read_file, _write_file, _file_str_replace, _workspace_grep, _compress_memory, _todo_write, _task_finish, _ask_user, _bash, _read_image, _create_sub_task, _batch_sub_tasks.

Custom tools can be registered via ToolRegistry.register().

Sub-agents

Configure named sub-agents on AgentTask.sub_agents. The parent agent delegates work via _create_sub_task / _batch_sub_tasks. Each sub-agent gets its own runtime, model, and tool set.

When a sub-agent uses a different model from the parent, the runtime needs settings_file and default_backend to resolve the LLM client.

Examples

24 numbered examples in examples/. See examples/README.md for the full list.

uv run python examples/01_quick_start.py
uv run python examples/24_workspace_backends.py

Testing

uv run pytest                              # unit tests (no network)
uv run ruff check .                        # lint
uv run ty check                            # type check

V_AGENT_RUN_LIVE_TESTS=1 uv run pytest -m live   # integration tests (needs real LLM)

Environment variables for live tests:

Variable	Default	Description
V_AGENT_LOCAL_SETTINGS	local_settings.py	Settings file path
V_AGENT_LIVE_BACKEND	moonshot	LLM backend
V_AGENT_LIVE_MODEL	kimi-k2.5	Model name
V_AGENT_ENABLE_BASE64_KEY_DECODE	-	Set 1 to enable base64 API key decoding