Proca

Agent runtime management built on Ray + LangGraph.

Proca treats LLM agents as processes managed by a lightweight kernel. It provides process lifecycle management, hierarchical memory with TTL/quota, semaphore-based scheduling, and streaming event propagation — all backed by Ray Actors for true parallelism.

Architecture

┌────────────────────────────────────────────────────────┐
│                     Consumer (your app)                │
│  WorkflowBase / FastAPI / CLI                          │
│  ┌──────────┐  ┌──────────┐  ┌────────────────┐        │
│  │  Tools   │  │  SSE     │  │  API Routes    │        │
│  └────┬─────┘  └────┬─────┘  └───────┬────────┘        │
│       │             │                │                 │
├───────┴─────────────┴────────────────┴─────────────────┤
│                       Proca Layer                      │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │ProcessManager│  │ MemoryStore  │  │AgentScheduler│  │
│  │  spawn/wait  │  │  ns/TTL/COW  │  │  semaphore   │  │
│  │  messaging   │  │              │  │              │  │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘  │
│         │                 │                 │          │
│  ┌──────┴─────────────────┴─────────────────┴───────┐  │
│  │              RayAgentActor (per agent)           │  │
│  │  LangGraph create_react_agent + astream          │  │
│  │  asyncio.Event cancel  ·  event queue            │  │
│  └──────────────────────────────────────────────────┘  │
│                                                        │
│  ┌──────────────────────────────────────────────────┐  │
│  │  ProcessManagerProxy (optional, cross-process)   │  │
│  │  _PMBridge Ray Actor  ·  serializable handle     │  │
│  └──────────────────────────────────────────────────┘  │
├────────────────────────────────────────────────────────┤
│  Ray Cluster (local or distributed)                    │
│  LangGraph BaseStore (PostgreSQL / InMemory)           │
└────────────────────────────────────────────────────────┘

Installation

# From the monorepo root
pip install -e proca/

# Or with dev dependencies
pip install -e "proca/[dev]"

Dependencies: ray[default]>=2.9.0, langgraph>=0.2.0, langchain-core>=0.3.0, pydantic>=2.0

Quickstart

import asyncio
import ray
from langchain_openai import ChatOpenAI
from langgraph.store.memory import InMemoryStore

from proca import ProcessManager, MemoryStore, AgentScheduler, SpawnConfig

async def main():
    ray.init()

    llm = ChatOpenAI(model="gpt-4o-mini")
    store = InMemoryStore()

    pm = ProcessManager(
        llm=llm,
        tools=[],                          # your LangChain tools
        store=store,
        system_prompt="You are a helpful analyst.",
        scheduler=AgentScheduler(max_concurrent=5),
        memory_store=MemoryStore(store),
    )

    # Spawn an agent
    agent_id = await pm.spawn(SpawnConfig(task="Summarize today's market"))

    # Wait for result
    result = await pm.wait(agent_id)
    print(result.text)

    ray.shutdown()

asyncio.run(main())

Core Concepts

1. ProcessManager — The Kernel

ProcessManager is the central coordinator. It creates Ray Actors, tracks parent-child relationships, and manages the full agent lifecycle.

from proca import ProcessManager, SpawnConfig

pm = ProcessManager(
    llm=llm,
    tools=tools,
    store=store,
    graph_factory=my_custom_graph_factory,  # optional: override default LangGraph creation
    max_depth=3,                            # max nesting depth for recursive agents
)

# Spawn
agent_id = await pm.spawn(SpawnConfig(task="Analyze NVDA"))

# Spawn multiple in parallel (uses asyncio.gather internally)
ids = await pm.spawn_parallel([
    SpawnConfig(task="Technical analysis"),
    SpawnConfig(task="Fundamental analysis"),
    SpawnConfig(task="Sentiment analysis"),
])

# Wait
result = await pm.wait(agent_id)           # single
results = await pm.wait_all(ids)           # all (parallel via asyncio.gather)
aid, result = await pm.wait_any(ids)       # first to finish

# Cancel (recursive — kills all children, non-blocking)
cancelled_ids = pm.cancel(agent_id)

# Query
handle = pm.get(agent_id)                  # single handle
active = pm.list_active()                  # running agents
tree = pm.get_tree()                       # full tree
stats = pm.get_stats()                     # summary

graph_factory

By default, ProcessManager uses langgraph.prebuilt.create_react_agent to build the agent graph. You can override this by passing a graph_factory callable:

def my_factory(llm, tools, *, store=None, **kwargs):
    """Custom graph factory — must return an object with .astream()."""
    return create_react_agent(llm, tools, store=store, ...)

pm = ProcessManager(llm=llm, tools=tools, graph_factory=my_factory)

This is especially useful for testing (inject mock graphs) and for customizing the agent's LangGraph topology.

2. MemoryStore — Namespace-Aware Memory

Wraps LangGraph's BaseStore with namespace hierarchy, TTL, quota, and COW snapshots.

Namespace hierarchy:
  agent_memory/shared/              ← all agents can read/write
  agent_memory/agent/{id}/          ← private to one agent
  agent_memory/agent/{id}/scratch/  ← ephemeral, auto-cleaned

from proca import MemoryStore

ms = MemoryStore(store, default_quota=500, default_ttl=3600)

# Shared memory
await ms.write_shared("market_trend", {"direction": "bullish"})
items = await ms.read_shared(query="market")

# Per-agent memory
await ms.write_agent("agent_abc", "analysis", {"result": "buy"})
items = await ms.read_agent("agent_abc")

# TTL: auto-expires after 60 seconds
await ms.put(("custom", "ns"), "temp_key", "data", ttl=60)

# Quota: max 100 items in this namespace
ms.set_quota(("custom", "ns"), 100)

# COW snapshot: copy parent memory to child
await ms.snapshot(
    src_namespace=("agent_memory", "agent", "parent_id"),
    dst_namespace=("agent_memory", "agent", "child_id"),
)

# Cleanup
await ms.cleanup_expired()                 # delete all expired items
await ms.cleanup_agent("agent_abc")        # delete agent's private + scratch

3. AgentScheduler — Concurrency Control

Semaphore + priority queue. Agents wait for a slot instead of being truncated.

from proca import AgentScheduler

sched = AgentScheduler(max_concurrent=5, max_pending=50)

# Context manager (recommended)
async with sched.slot("agent_id", priority=10):
    await run_agent(...)

# Manual
await sched.acquire("agent_id", priority=10)
try:
    await run_agent(...)
finally:
    sched.release("agent_id")

# Query
print(sched.running)      # currently running
print(sched.pending)      # waiting in queue
print(sched.available)    # free slots
print(sched.get_stats())  # full stats

Priority: higher value = scheduled sooner. Equal priority = FIFO.

4. RayAgentActor — The Worker

Each agent runs as an async Ray Actor wrapping a LangGraph ReAct agent. You usually don't interact with this directly — ProcessManager handles it.

from proca.actor import RayAgentActor

actor = RayAgentActor.remote(
    agent_id="agent_001",
    llm=llm,
    tools=tools,
    system_prompt="You are helpful.",
    store=store,
    config={"timeout": 600, "max_iterations": 100},
    parent_id=None,
    graph_factory=my_factory,       # optional: override default graph creation
)

# Run (returns AgentResult)
result_ref = actor.run.remote("Analyze the market")
result = ray.get(result_ref)

# Stream events
events = ray.get(actor.drain_events.remote())

# Cancel
ray.get(actor.cancel.remote())

5. Events — Streaming Protocol

Agents emit AgentEvent objects during execution. These are serializable dicts suitable for SSE/WebSocket transport. Event timestamps use UNIX wall-clock time (time.time()) for cross-process comparability.

from proca.events import EventType

# Event types:
# Lifecycle:  AGENT_STARTED, AGENT_COMPLETED, AGENT_FAILED, AGENT_CANCELLED, AGENT_TIMEOUT
# ReAct:      THINKING_START, THINKING_TOKEN, THINKING_END, TOOL_CALL_START, TOOL_CALL_END
# Spawn:      SPAWN_CHILD, CHILD_COMPLETED
# Memory:     MEMORY_WRITE, MEMORY_READ

# Drain events from all running agents
events = await pm.drain_all_events()
for ev in events:
    print(ev["event_type"], ev["agent_id"], ev["data"])

6. Tools — LangChain Integration

create_proca_tools() generates @tool functions that agents can use to spawn children, manage memory, and query the process tree.

Same-process usage (driver process)

When tools run in the same process as ProcessManager (e.g. a top-level LangGraph workflow in the driver):

from proca.tools import create_proca_tools

tools = create_proca_tools(
    process_manager=pm,
    memory_store=ms,
    current_agent_id="root_agent",
    current_depth=0,
    max_depth=3,
)
# Returns: [spawn_subagent, spawn_parallel_subagents, wait_for_agent,
#           check_agent_status, list_agents, cancel_agent,
#           send_message, check_messages,
#           read_shared_memory, write_shared_memory,
#           read_own_memory, write_own_memory]

Cross-process usage (tools inside Ray Actors)

When tools need to run inside Ray worker processes (e.g. agents spawning sub-agents), use ProcessManagerProxy — a Ray-serializable handle:

from proca import ProcessManagerProxy
from proca.tools import create_proca_tools

# Create a remote PM (all state lives in a Ray actor):
proxy = ProcessManagerProxy.create(
    llm=llm,
    tools=user_tools,
    store=store,
    graph_factory=my_factory,
    max_concurrent=5,
    memory_quota=500,
)

# Create cross-process-safe tools (proxy is serializable by Ray):
proca_tools = create_proca_tools(
    process_manager=proxy,      # no memory_store needed — proxy handles it
    current_agent_id="root",
    current_depth=0,
    max_depth=3,
)

# These tools can be safely passed to RayAgentActor or any Ray worker

The proxy and local PM share the same interface. create_proca_tools detects which one it receives and routes memory operations accordingly.

Depth control: when current_depth >= max_depth, spawn tools are excluded automatically.

7. SpawnConfig — Agent Configuration

from proca import SpawnConfig

config = SpawnConfig(
    task="Analyze NVDA technical indicators",  # natural language task
    timeout=600.0,                             # max seconds (default 600)
    max_iterations=100,                        # LangGraph recursion limit
    priority=10,                               # higher = scheduled first
    memory_inherit=True,                       # COW-copy parent memory
    system_prompt_override="You are a TA expert.",
    tools_filter=["get_market_data", "execute_python"],  # tool allowlist
    parent_id="parent_agent_id",               # set by ProcessManager
    depth=1,                                   # set by ProcessManager
    # Per-spawn overrides (enable heterogeneous agent trees)
    llm_override=ChatOpenAI(model="gpt-4o"),   # use a different LLM for this agent
    extra_tools=[my_custom_tool],              # additional tools (merged with base)
    graph_factory_override=my_factory,         # custom graph factory for this agent
)

Per-spawn overrides

Each SpawnConfig can override the default LLM, tools, and graph factory set on ProcessManager:

Field	Default	Description
llm_override	None (use PM default)	Use a different LLM for this specific agent
extra_tools	[]	Additional tools appended to the base tool list
graph_factory_override	None (use PM default)	Custom graph factory for this agent's topology

This enables heterogeneous agent trees — e.g. a coordinator using GPT-4o that spawns workers using GPT-4o-mini, or specialists with different tool sets.

8. AgentResult — Return Value

result = await pm.wait(agent_id)

result.agent_id      # "agent_abc123def"
result.status        # AgentStatus.COMPLETED / FAILED / CANCELLED / TIMEOUT
result.text          # final LLM output
result.tool_calls    # ["get_market_data", "execute_python"]
result.tool_results  # [{"tool_call_id": "...", "content": "..."}]
result.duration_ms   # 4523
result.exit_code     # 0=success, 1=cancelled, 2=timeout, 3=error
result.error         # None or error message
result.success       # True if exit_code == 0

9. ProcessManagerProxy — Cross-Process Access

When agents need to spawn sub-agents (recursive agent trees) or access memory from within Ray workers, use ProcessManagerProxy:

from proca import ProcessManagerProxy

# Create (all state lives inside a _PMBridge Ray actor):
proxy = ProcessManagerProxy.create(
    llm=llm,
    tools=tools,
    store=store,
    system_prompt="You are a helpful analyst.",
    max_concurrent=5,
    max_pending=50,
    memory_quota=500,
    memory_ttl=3600,
    graph_factory=my_factory,
    max_depth=3,          # max nesting depth for recursive agents
)

# Same async API as ProcessManager:
agent_id = await proxy.spawn(SpawnConfig(task="Analyze NVDA"))
result = await proxy.wait(agent_id)

# Memory operations (no separate MemoryStore needed):
await proxy.memory_write_shared("key", {"data": "value"})
items = await proxy.memory_read_shared()

# Messaging:
await proxy.send_message("agent_a", "agent_b", "hello")
msgs = await proxy.recv_messages("agent_b")

# Query:
handle = await proxy.get(agent_id)    # returns dict (not AgentHandle)
tree = await proxy.get_tree()
stats = await proxy.get_stats()

Note: ProcessManagerProxy methods are all async (including get, cancel, get_tree), since they perform Ray remote calls. The local ProcessManager keeps its synchronous query methods for convenience.

Auto-inject proca tools

When using ProcessManagerProxy.create(), child agents automatically receive proca tools (spawn, wait, memory, messaging) at spawn time. This is the key enabler for recursive agent trees — no manual tool wiring needed.

• At depth < max_depth: agents get all proca tools including spawn_subagent
• At depth == max_depth: agents get query/wait/messaging tools but NOT spawn tools
• Without a proxy: no tools are auto-injected (you wire them manually)

proxy = ProcessManagerProxy.create(llm=llm, tools=user_tools, max_depth=3)

# Root agent (depth=0) gets spawn + wait + memory + messaging tools auto-injected
root_id = await proxy.spawn(SpawnConfig(task="Coordinate analysis"))

# If the root agent calls spawn_subagent, the child (depth=1) also gets tools
# Children at depth=3 can wait/message but not spawn further

10. Inter-Agent Messaging

Agents can communicate directly via a built-in mailbox system in ProcessManager:

# Same-process (local PM)
pm.send_message(from_id="agent_a", to_id="agent_b", content="Analysis done")
msgs = pm.recv_messages("agent_b")  # drain semantics: messages removed after read

# Cross-process (via proxy)
await proxy.send_message("agent_a", "agent_b", "Analysis done")
msgs = await proxy.recv_messages("agent_b")

Agents have access to messaging via auto-injected tools:

Tool	Description
send_message(target_agent_id, message)	Send a message to another agent's inbox
check_messages(limit=10)	Read & drain messages from own inbox

Messages are simple dicts with from_agent_id, content, and timestamp fields. Drain semantics: once read, messages are removed from the inbox.

Exception Hierarchy

ProcaError
├── AgentNotFoundError(agent_id)
├── AgentAlreadyExistsError(agent_id)
├── AgentCancelledError(agent_id)
├── AgentTimeoutError(agent_id, timeout)
├── MemoryQuotaExceededError(namespace, quota)
└── SchedulerFullError(max_pending)

Testing

# Run all tests
cd proca/
pytest tests/ -v

# Run without Ray-dependent tests (faster)
pytest tests/ -v --ignore=tests/test_actor.py --ignore=tests/test_integration.py

Tests use a real single-node Ray cluster (not local_mode, which doesn't support async actors). Mock graphs are injected via the graph_factory parameter — no monkey-patching needed.

Key test areas:

• Per-spawn overrides: llm_override, graph_factory_override, extra_tools via SpawnConfig
• Auto-inject: proca tools injected/omitted correctly based on proxy and depth
• Messaging: send/recv isolation, drain semantics, cross-agent routing
• Recursive agents: end-to-end with ProcessManagerProxy

Examples

See the examples/ directory:

File	Description
01_basic_spawn.py	Spawn a single agent, wait for result
02_parallel_analysis.py	Multi-agent parallel analysis
03_memory_sharing.py	Shared + private memory between agents
04_agent_tree.py	Parent-child tree with recursive cancel
05_scheduler_priority.py	Semaphore concurrency + priority scheduling

Integration with Agent-Trader

Proca is designed as a standalone library. To integrate with Agent-Trader:

1. subagent.py → delegates to ProcessManager.spawn() instead of manual LangGraph creation
2. system_tools.py → imports tools from proca.tools instead of inline definitions
3. agent_registry.py → becomes a thin adapter over ProcessManager
4. workflow_base.py → initializes ProcessManager with the workflow's LLM, store, and tools
5. For recursive agent spawning across Ray workers, use ProcessManagerProxy.create() and pass the proxy to create_proca_tools()