agentorch

Code-first · Async-first Python framework for multi-agent orchestration

English | 简体中文 | 繁體中文 | Français | 日本語 | 한국어 | Español

An explicit runtime for real-world agent systems: compose models, tools, retrieval, memory, workflows, and delegation into observable software.

The PyPI distribution is masarch; install it with pip install masarch and import it as agentorch.

Project Overview

agentorch is for teams and researchers who need controllable agent systems rather than hidden prompt pipelines. It provides explicit runtime objects, policy boundaries, and structured state so a system can grow from one agent into coordinated multi-agent execution.

Typical Scenarios

• Coding assistants with bounded filesystem, shell, Git, and review access.
• Knowledge assistants with retrieval evidence, RAG, citations, and explainable output.
• Workflow automation with DAG execution, retries, audit trails, and persistence.
• Long-running tasks with threads, workspaces, cross-turn memory, and resumability.

Repository Layout

Directory	Contents
agentorch/	Core runtime, models, tools, memory, and workflow components
projects/	Application projects such as AI text detection and AI short drama
experiments/	Long-term memory, context, and baseline experiments
resource/	Architecture diagrams, runtime flows, and brand assets
tests/	Cross-module integration and project tests
data/	Local data contract; raw data is ignored by default

Technology Stack

Layer	Technology	Purpose
Core language	Python 3.10+, asyncio	Async-first runtime and task scheduling
Types and validation	Pydantic	Validate requests, responses, tools, and configuration
Model adapters	OpenAI API and compatible endpoints	Chat, embeddings, vision, and speech models
Orchestration	Runtime and Workflow DAG	Routing, delegation, handoffs, retries, and state
Tooling	filesystem, execution, git, web, media	Policy-controlled tool execution
Knowledge and memory	RAG, Memory, optional Neo4j	Evidence retrieval, long-term memory, and graphs
Engineering	pytest, Jinja2, HTTPX	Testing, generation templates, and HTTP adapters

Core Capabilities

• Model adapters, structured output, and streaming responses
• Tool registries, allowlists, sandboxes, and permission policies
• Reasoning strategies such as react and plan_execute
• Multi-agent delegation, task packets, and explicit handoff records
• RAG retrieval, evidence mounting, memory retention, and promotion
• Workflow execution, runtime tracing, and observability events

Architecture Overview

Runtime Flow

Architecture diagrams, technology icons, and project logos live in resource/brand.

Datasets, databases, model files, and generated experiment artifacts are ignored by default. See data/README.md for the sharing contract.

Source repository: https://github.com/Akun-python/agentorch

WHY

Why This Project Exists 🎯

Many projects hit a wall after the "single assistant + one prompt" phase.

The moment you need specialist roles, constrained tools, repeatable state, and observable handoffs, ad-hoc prompt glue becomes difficult to reason about.

agentorch is designed to keep these concerns explicit:

• model adapter choices
• tool exposure and safety boundaries
• retrieval strategy and evidence mounting
• memory retention and promotion
• workflow execution order
• multi-agent coordination and delegation

Why It Helps Engineering Teams 🧭

• You can inspect system assembly with exported blueprint/config.
• You can enforce policy boundaries with typed configs.
• You can evolve behavior (reasoning/RAG/workflow) without rewriting everything.
• You can test behavior through code-level contracts.

Why It Helps Research Teams 🔬

• Swappable reasoning modes (react, plan_execute, etc.)
• Search/evolution support for strategy comparison
• Source-aware RAG flow and evidence-oriented outputs
• Long-horizon memory patterns for iterative tasks

Typical Scenarios

• multi-agent coding assistants with bounded filesystem/shell access
• research copilots that must cite retrieved sources
• workflow-driven automation that needs deterministic node execution
• long-running assistants with thread/workspace memory

WHAT

Core Facade API

• create_agent(...)
• create_multi_agent(...)

These are the recommended entrypoints for most users.

Key Runtime Building Blocks 🧩

• model adapters (OpenAIModel, compatible HTTP adapters)
• tool registry and bundles
• sandbox manager and policy
• knowledge base and RAG strategy
• memory manager and memory policy
• workflow DAG builder and runner
• observability hooks and SQLite event store

Built-In Capability Surface

• structured tool calling via Pydantic I/O
• filesystem / execution / git / web / media bundles
• multi-format ingestion (md, txt, pdf, docx, code artifacts)
• reasoning strategy selection
• human feedback and resumable flows
• extension hooks for lifecycle interception

What "Orchestration" Means Here

In agentorch, orchestration is not a marketing word.

It means each runtime concern has a concrete type and place in assembly:

• coordinator policies decide routing behavior
• supervisor plans are inspectable objects
• handoffs and task packets are explicit records
• memory scopes and shared state are controlled by policy

Compatibility and Stability

• Python 3.10+
• minimal core dependencies
• stable high-level facade surface for day-to-day use
• compatibility exports for older integrations

HOW

Installation 📦

Install from PyPI:

pip install masarch

If your package mirror has not synchronized the latest release yet, use the official PyPI index:

pip install -i https://pypi.org/simple --no-cache-dir masarch

Verify the installed distribution and import package:

python -c "import importlib.metadata as m; import agentorch; print(m.version('masarch')); print(agentorch.__file__)"

Query release versions:

pip index versions masarch -i https://pypi.org/simple

Local editable install:

pip install -e .

Direct install from GitHub:

pip install "git+https://github.com/Akun-python/agentorch.git"

Optional extras example:

pip install "masarch[neo4j]"

Local editable install with optional extras:

pip install -e ".[neo4j]"

Environment Setup

Set provider credentials through environment variables:

OPENAI_API_KEY=sk-xxxx
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_EMBEDDING_MODEL=text-embedding-3-small

Local .env loading is opt-in.

Recommended Start Path

1. Start with create_agent(...) and one minimal tool.
2. Add RAG only after baseline behavior is stable.
3. Add workflow DAG only when execution order matters.
4. Move to create_multi_agent(...) when role separation is clear.

Validation Commands

Run package tests:

py -3.10 -m pytest -q

Run README contract tests:

py -3.10 -m pytest -q agentorch/tests/test_readme_contracts.py

Practical Guardrails ✅

• keep tool allowlists narrow
• avoid enabling shell where not required
• keep thread IDs explicit for traceability
• close agents/runtimes after use

QUICKSTART

1) Minimal Agent (sync)

from agentorch import create_agent

agent = create_agent(
    model="gpt-4.1-mini",
    system_prompt="You are concise and accurate.",
    reasoning="react",
)

result = agent.run_sync(
    "Explain what agent orchestration is in three bullet points.",
    thread_id="quickstart-en-001",
)

print(result.output_text)
agent.close()

2) Tool Calling

from pydantic import BaseModel

from agentorch import ToolRegistry, create_agent, tool

class AddInput(BaseModel):
    a: int
    b: int

@tool(description="Add two integers.")
async def add_numbers(input: AddInput):
    return {"sum": input.a + input.b}

agent = create_agent(
    model="gpt-4.1-mini",
    tools=ToolRegistry.from_tools(add_numbers),
    reasoning="react",
)

result = agent.run_sync("Use add_numbers to compute 12 + 30.", thread_id="quickstart-tools-001")
print(result.output_text)
agent.close()

3) Multi-Agent Starter

from agentorch import create_agent, create_multi_agent

planner = create_agent(model="gpt-4.1-mini", reasoning="plan_execute", name="planner")
reviewer = create_agent(model="gpt-4.1-mini", reasoning="react", name="reviewer")

team = create_multi_agent(
    model="gpt-4.1-mini",
    agents=[
        {"agent": planner, "name": "planner", "role": "planner"},
        {"agent": reviewer, "name": "reviewer", "role": "reviewer"},
    ],
    system_prompt="Coordinate specialists and return one final answer.",
)

result = team.run_sync("Draft and review a migration plan.", thread_id="quickstart-team-001")
print(result.output_text)
team.close()

4) Next Steps

The recommended path from a prototype to a maintainable system is to make one new boundary explicit at a time:

A. Mount project knowledge with RAG

Use knowledge_paths for local Markdown, text, code, or supported documents. enable_rag=True makes retrieval part of the runtime contract instead of leaving it to prompt assembly. Keep the source directory small, versioned, and safe to share; do not point it at raw datasets or generated artifacts.

from agentorch import create_agent

agent = create_agent(
    model="gpt-4.1-mini",
    enable_rag=True,
    knowledge_paths=["docs/", "src/"],
    knowledge_scope=["public-docs", "source-code"],
)
result = agent.run_sync("根据项目文档解释认证流程。", thread_id="rag-001")
print(result.output_text)
agent.close()

B. Make important steps deterministic with a workflow DAG

Use a workflow when the order, retry boundary, approval point, or handoff must be inspectable. Keep open-ended reasoning inside a model node, and use explicit tool, retrieval, approval, and aggregation nodes for system boundaries.

from agentorch import WorkflowBuilder, create_agent
from agentorch.workflow import Node

workflow = (
    WorkflowBuilder(max_steps=8)
    .add(Node.retrieve("retrieve", question="用户问题"), entry=True)
    .then(Node.model_node("draft", prompt="基于检索证据形成初稿。"))
    .then(Node.model_node("review", prompt="检查事实、引用和风险。"))
    .build()
)
agent = create_agent(model="gpt-4.1-mini", workflow=workflow)
result = agent.run_sync("整理本次变更", thread_id="workflow-001")
agent.close()

C. Turn on observability before cost or quality tuning

Observability records runtime events, usage, handoffs, and failure context in a local SQLite store. Use a project-local path, redact secrets, and compare traces by model, prompt version, workflow version, and thread type.

from agentorch import create_agent
from agentorch.config import ObservabilityConfig

agent = create_agent(
    model="gpt-4.1-mini",
    observability=ObservabilityConfig(
        enabled=True,
        sqlite_path=".agentorch/observability.db",
        console_mode="important_only",
    ),
)

Track at least: success rate, tool failure rate, latency, prompt/completion tokens, estimated cost, retrieval hit quality, citation coverage, and human fallback rate. Do not put API keys, raw personal data, or datasets into trace payloads.

D. Encode team boundaries with policy objects

Policies are executable configuration. Pin them in code or a reviewed config file so a multi-agent team has predictable context, state, routing, memory, and tool behavior.

from agentorch import (
    ContextPolicy,
    CoordinationPolicy,
    MemoryPolicy,
    StatePolicy,
    create_multi_agent,
)

team = create_multi_agent(
    model="gpt-4.1-mini",
    roles=[
        {"name": "planner", "description": "Plans work", "capabilities": ["plan"]},
        {"name": "reviewer", "description": "Reviews work", "capabilities": ["review"]},
    ],
    context_policy=ContextPolicy.evidence_friendly(),
    state_policy=StatePolicy(retention_mode="state_plus_memory"),
    coordination_policy=CoordinationPolicy.distributed(),
    memory_policy=MemoryPolicy.long_horizon(),
)

Start with least privilege: narrow tool allowlists, explicit knowledge scopes, bounded delegation depth, and summary_only handoffs. Only widen a policy when an evaluation demonstrates a real benefit.

E. Close the production loop

• Add unit and contract tests for adapters, tools, policies, and workflow edges.
• Add end-to-end tests with a mock provider before spending on live models.
• Maintain a small, versioned evaluation set for task success, grounding, safety, latency, and cost; pin model and prompt versions during comparisons.
• Persist thread IDs and workflow versions so failed runs can be reproduced or resumed without uploading local data.
• Add human approval nodes for destructive actions, external messages, and high-risk tool calls.
• Deploy the runtime with bounded concurrency, timeouts, retries, rate limits, health checks, and a clear shutdown path.

Quick FAQ

Q: Should I start with multi-agent first?
A: Usually no. Start with one strong agent, then split roles when boundaries are clear.

Q: When should I enable workflow DAG?
A: When task order matters and you want deterministic step execution.

Q: When should I enable long-term memory?
A: When tasks span multiple threads/sessions and prior outputs must be reused.

Q: How do I keep tool execution safe?
A: Use sandbox policy, strict allowlists, and narrow workspace scopes.

Troubleshooting Notes 🛟

• TypeError around modern typing syntax usually means Python version is too low.
• If python points to an older interpreter, use explicit launcher command (py -3.10).
• If output feels unstable, pin model version and keep thread IDs consistent.
• If delegation is noisy, reduce agent count and tighten role descriptions first.

Reference Entry Points

• Main docs: README.md (this file)
• Simplified Chinese: README.zh-CN.md
• Examples folder: examples/
• Package tests: agentorch/tests/

For production usage, treat this README as a launch map and move critical settings into versioned config files.

MIT License.