Migrating from AWorld to Exo

from aworld.agents import LLMAgent
from aworld.config.conf import AgentConfig, TaskConfig

agent_config = AgentConfig(
    name="my-agent",
    llm_provider="openai",
    llm_model_id="gpt-4o",
    max_steps=10,
    temperature=0.7,
)
task_config = TaskConfig(name="my-task", description="Do something")

agent = LLMAgent(agent_config=agent_config, task_config=task_config)

from exo import Agent

agent = Agent(
    name="my-agent",
    model="openai:gpt-4o",        # provider:model in one string
    instructions="Do something",    # instructions replace task_config description
    max_steps=10,
    temperature=0.7,
)

# ConfigDict (dict subclass with attribute access)
from aworld.config.conf import ConfigDict
config = ConfigDict({"name": "agent", "llm_provider": "openai"})
print(config.name)  # attribute access on a dict

# BaseConfig (custom base class)
from aworld.config.conf import BaseConfig
class MyConfig(BaseConfig):
    name: str
    value: int = 0

# Pydantic models (some places)
from pydantic import BaseModel
class AgentConfig(BaseModel):
    name: str

from exo.config import AgentConfig, ModelConfig, TaskConfig, RunConfig

# All configs are frozen Pydantic models
agent_config = AgentConfig(
    name="researcher",
    model="openai:gpt-4o",
    instructions="Research topics thoroughly.",
    temperature=0.7,
    max_steps=20,
)

model_config = ModelConfig(
    provider="openai",
    model_name="gpt-4o",
    api_key="sk-...",
    timeout=30.0,
)

from aworld.tools.function_tools import FunctionTool
from aworld.core.tool.base import BaseTool, AsyncBaseTool

# Function tool
tool = FunctionTool(name="search", func=search_fn, description="Search the web")

# Class-based tool (returns gym-style 5-tuple)
class MyTool(BaseTool):
    def execute(self, action):
        return observation, reward, terminated, truncated, info

# Async version was a separate class
class MyAsyncTool(AsyncBaseTool):
    async def execute(self, action):
        return observation, reward, terminated, truncated, info

from exo import tool, Tool

# Decorator (preferred) -- sync or async
@tool
def search(query: str) -> str:
    """Search the web and return results."""
    return "results"

@tool
async def fetch_url(url: str) -> str:
    """Fetch content from a URL."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.text

# Class-based (for complex tools)
class DatabaseQuery(Tool):
    name = "query_database"
    description = "Execute a SQL query."
    parameters = {
        "type": "object",
        "properties": {
            "sql": {"type": "string", "description": "The SQL query"},
        },
        "required": ["sql"],
    }

    async def execute(self, **kwargs) -> str:
        sql = kwargs["sql"]
        return await db.execute(sql)

from aworld.runner import create_runner

# 6 different entry points
runner = create_runner(agent_config=config, task_config=task_config)
result = await runner.run(task)
result = runner.sync_run(task)
async for event in runner.streaming_run(task): ...
result = await runner.run_task(task)
async for event in runner.streaming_run_task(task): ...
result = await runner.streamed_run_task(task)

from exo import Agent, run

agent = Agent(name="assistant", model="openai:gpt-4o", instructions="Be helpful.")

# 3 entry points
result = await run(agent, "What is 2+2?")                    # async
result = run.sync(agent, "What is 2+2?")                      # sync
async for event in run.stream(agent, "Tell me a story"):       # streaming
    if event.type == "text":
        print(event.text, end="", flush=True)
    elif event.type == "tool_call":
        print(f"\n[calling {event.tool_name}...]")

# Multi-turn conversation
result = await run(agent, "What's the weather?")
result = await run(agent, "What about tomorrow?", messages=result.messages)

from aworld.agents import SwarmComposerAgent
from aworld.agents.swarm_builder import SwarmBuilder

# Builder pattern
builder = SwarmBuilder()
builder.add_agent(researcher)
builder.add_agent(writer)
builder.add_agent(editor)
builder.add_edge(researcher, writer)
builder.add_edge(writer, editor)
swarm = builder.build()

from exo import Agent, Swarm, run

researcher = Agent(name="researcher", ...)
writer = Agent(name="writer", ...)
editor = Agent(name="editor", ...)

# Sequential pipeline with flow DSL
pipeline = Swarm(
    agents=[researcher, writer, editor],
    flow="researcher >> writer >> editor",
)
result = await run(pipeline, "Write an article about quantum computing")

# Handoff (agent-driven delegation)
triage = Agent(name="triage", handoffs=[billing, support], ...)
swarm = Swarm(agents=[triage, billing, support], mode="handoff")
result = await run(swarm, "I need a refund")

# Team (lead + workers)
team = Swarm(agents=[lead, analyst, writer], mode="team")
result = await run(team, "Research and report on market trends")

# Parallel groups in flow
parallel_pipeline = Swarm(
    agents=[fetcher, analyzer_a, analyzer_b, summarizer],
    flow="fetcher >> (analyzer_a | analyzer_b) >> summarizer",
)

from aworld.core.context.amni.contexts import AmniContext
from aworld.core.context.amni.config import AmniConfig, AmniConfigFactory
from aworld.core.context.amni.prompt.prompts import PromptService
from aworld.core.context.amni.prompt.neurons.neuron_factory import NeuronFactory

# Factory-based config
config = AmniConfigFactory.create("copilot")
context = AmniContext(config=config)

# Neuron factory registration
NeuronFactory.register("task", TaskNeuron)
NeuronFactory.register("history", HistoryNeuron)

# Service-based prompt building
prompt_service = PromptService(context)
prompt = await prompt_service.build_prompt(neuron_names=["task", "history"])

from exo.context import Context, ContextConfig, PromptBuilder

# Config-based (no factory)
config = ContextConfig(
    mode="copilot",            # "pilot" | "copilot" | "navigator"
    history_rounds=20,
    summary_threshold=30,
    offload_threshold=4000,
)
ctx = Context(task_id="task-1", config=config)

# Direct prompt building (no factory)
builder = PromptBuilder(ctx)
builder.add("system", "You are a helpful assistant.")
builder.add("task")
builder.add("history")
builder.add("knowledge", query="quantum computing")
prompt = await builder.build()

# Hierarchical state
child_ctx = ctx.fork(task_id="subtask-1")
child_ctx.state.set("progress", "researching")  # local only
ctx.merge(child_ctx)  # merge results back

from aworld.memory.main import Memory
from aworld.memory.longterm.default import DefaultLongTermMemory
from aworld.memory.embeddings.factory import EmbeddingFactory

memory = Memory(
    long_term=DefaultLongTermMemory(
        embedding=EmbeddingFactory.create("openai"),
        db_path="memory.db",
    )
)

from exo.memory import ShortTermMemory, LongTermMemory

agent = Agent(
    name="assistant",
    memory=LongTermMemory(backend="sqlite", path="memory.db"),
    ...
)