vv-agent 0.1.12

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

vv-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 vv_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 vv-agent --prompt "Summarize this framework" --backend moonshot --model kimi-k2.5

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

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

Programmatic

from vv_agent.config import build_openai_llm_from_local_settings
from vv_agent.runtime import AgentRuntime
from vv_agent.tools import build_default_registry
from vv_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 vv_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 vv_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 vv_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 vv_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 'vv-agent[s3]'.

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

Memory Compaction

MemoryManager compacts history when AgentTask.memory_compact_threshold is exceeded.

• Task-level knobs:
  • memory_compact_threshold (default 128000)
  • memory_threshold_percentage (warning threshold percentage, default 90)
• Effective-length strategy (backend-aligned):
  • If previous cycle token usage exists:
    • effective_length = previous_total_tokens + len(json.dumps(recent_tool_messages))
  • Otherwise fallback to:
    • len(json.dumps(messages[2:]))
• Compaction pipeline:
  1. Structural cleanup (stale tool calls, orphan tool messages, assistant-no-tool collapse, old tool result artifactization)
  2. If still over threshold, generate compressed memory summary

Runtime metadata keys

Pass these via AgentTask.metadata:

• memory_keep_recent_messages
• include_memory_warning
• tool_result_compact_threshold
• tool_result_keep_last
• tool_result_excerpt_head
• tool_result_excerpt_tail
• tool_calls_keep_last
• assistant_no_tool_keep_last
• tool_result_artifact_dir
• summary_event_limit

Memory summary model selection priority

Priority is strict:

1. AgentTask.metadata
   • memory_summary_backend / memory_summary_model
   • aliases: compress_memory_summary_backend / compress_memory_summary_model
   • aliases: memory_compress_backend / memory_compress_model
2. local_settings.py constants
   • DEFAULT_USER_MEMORY_SUMMARIZE_BACKEND / DEFAULT_USER_MEMORY_SUMMARIZE_MODEL
   • aliases: DEFAULT_MEMORY_SUMMARIZE_BACKEND / DEFAULT_MEMORY_SUMMARIZE_MODEL
   • aliases: VV_AGENT_MEMORY_SUMMARY_BACKEND / VV_AGENT_MEMORY_SUMMARY_MODEL
3. Fallback
   • runtime default_backend + current task model

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