orch-lib

A Python library for agent orchestration with task scheduling, conditional routing, and memory management.

Features

• Agent Orchestration: Build complex multi-agent workflows with central coordination
• Task Management: Define and execute tasks with memory scoping
• Conditional Routing: Route task execution based on dynamic conditions
• Memory Management: Isolated and shared memory across agents and graphs
• Graph Execution: Coordinate multiple agents at the graph level
• Expression Evaluation: Evaluate dynamic expressions with memory access

Installation

pip install orch-lib

Quick Start

Basic Agent Example

from orch_lib import Agent, Task, Memory

# Create an agent
agent = Agent(name="worker")

# Define a task
def process_data(scope=None):
    scope["private"]["result"] = "processed"
    return scope["private"]["result"]

task = Task("process", process_data)

# Add task to agent
agent.add_task("process", task)
agent.add_route("process", "on_start")

# Execute agent
agent()

Agent with Conditional Routing

from orch_lib import Agent, Task, Conditional, Memory

agent = Agent(memory=Memory())
agent.memory.store("should_run", True)

def task_fn(*args, scope=None):
    print("Task executed!")

agent.add_task("work", Task("work", task_fn))
agent.add_route("work", Conditional("private['should_run']"))

agent(max_passes=2)

Graph-Level Orchestration

from orch_lib import Graph, Agent, Task, Conditional

graph = Graph()

# Create agents
agent1 = Agent()
agent2 = Agent(parent=graph)

# Configure agents with tasks and routes
# ... (add tasks and routes as shown above)

# Add to graph
graph.add_agent("agent1", agent1)
graph.add_agent("agent2", agent2)

# Execute graph
graph()

Core Concepts

Memory

Agents and graphs maintain isolated memory spaces:

• Private Memory: Agent-level memory not accessible to other agents
• Public Memory: Graph-level memory accessible to all agents
• Agent Memory: Per-agent memory in graph context

Tasks

Tasks are callable units of work:

task = Task(
    name="my_task",
    function=my_function,
    memories={"key": "value"}  # task-level memory
)

Conditionals

Routes can be conditional:

condition = Conditional("private['flag'] and public['enabled']")
agent.add_route("task_id", condition)

Expressions

Evaluate expressions with memory context:

from orch_lib import Expression

expr = Expression("count + 1")
result = expr(parent={"count": 5})

API Reference

Agent

Agent(
    tasks=None,           # dict of Task objects
    memory=None,          # Memory object
    routes=None,          # dict of routes
    name=None,            # agent name
    parent=None,          # parent memory reference
    max_passes=10000      # max execution passes
)

Task

Task(
    name: str,
    function: Callable,
    memories: dict = None
)

Conditional

Conditional(
    condition: str,       # Python expression string
    parent=None           # parent memory reference
)

Memory

Memory(agents=None)

Methods: store(), retrieve(), delete(), clear(), clear_agent_memory()

Graph

Graph(
    agents=None,          # dict of Agent objects
    memory=None,          # Memory object
    routes=None,          # dict of conditional routes
    max_passes=10000      # max execution passes
)

Development

Developer Docs

• Full developer guide: docs/DEVELOPER_GUIDE.md
• Converter DSL reference: docs/CONVERTER_DSL_REFERENCE.md

Setup

pip install -e ".[dev]"

Running Tests

pytest

Code Quality

black src/
ruff check src/
mypy src/

License

MIT License - see LICENSE file for details

Contributing

Contributions are welcome! Please open an issue or submit a pull request.