Agent

The Agent is the autonomous actor in YosrAI. It encapsulates an LLM, a system persona, and a set of tools to perform tasks.

Purpose

Use the Agent when you want to: - Interact with an LLM using a specific persona. - Execute tools (functions) autonomously based on user input. - Maintain a conversation loop (ReAct pattern) to solve complex problems.

Usage

Basic Initialization

You can create an agent with just a name and instructions.

from yosrai import Agent

agent = Agent(
    name="Poet",
    instructions="You are a helpful poet. Write short haikus.",
    model="openai/gpt-4o"  # Provider/Model syntax
)

response = agent.run("Write a poem about coding.")
print(response)

Using Tools

Agents can be equipped with tools to perform actions.

from yosrai import Agent, tool

@tool
def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return f"The weather in {city} is sunny."

agent = Agent(
    name="WeatherBot",
    instructions="You provide weather updates.",
    tools=[get_weather]
)

agent.run("What's the weather in Tokyo?")

Async Support

YosrAI agents support asynchronous execution for high-throughput applications.

response = await agent.arun("Hello!")

Verbose Mode

Enable step-by-step reasoning output:

agent = Agent(
    name="Assistant",
    instructions="...",
    verbose=True  # Shows thinking process
)

Thinking Mode

Force the agent to think step-by-step using <thinking> tags:

agent = Agent(..., thinking=True)

Reflection Mode

Enable automatic self-critique and refinement:

agent = Agent(..., reflection=True)

Tracing & Replay

Capture execution for analysis:

result = agent.run("Hello", trace=True)
result.replay()  # Review with timing
trace_data = result.to_dict()  # Export as JSON

Blueprint System

Serialize and reconstruct agents:

# Export as JSON
blueprint = agent.to_blueprint()

# Reconstruct from JSON
reconstructed = Agent.from_blueprint(blueprint)

Code Export

Generate Python code:

code = agent.to_code()  # Complete agent definition

Conductor

The Conductor is a specialized Agent that orchestrates other agents or tools (called Skills).

Features

• Planning: Auto-generate execution plans.
• Shared Memory: Inject memory into all skills.
• Resilience: Wrap skills with SkillConfig.

Usage

from yosrai import Conductor, Agent, SkillConfig
from yosrai.engine.memory import LocalVectorMemory

# 1. Define Skills
writer = Agent(name="Writer", ...)
researcher = Agent(name="Researcher", ...)

# 2. Configure Resilience (Optional)
safe_researcher = SkillConfig(
    skill=researcher,
    retries=3,
    timeout=30.0
)

# 3. Create Shared Memory (Optional)
memory = LocalVectorMemory("team_brain.json", ...)

# 4. Initialize Conductor
conductor = Conductor(
    name="Manager",
    instructions="Coordinate the team.",
    skills=[safe_researcher, writer],
    planning=True,           # Enable planning
    shared_memory=memory     # Share memory
)

conductor.run("Research topic X and write a summary.")

Pipeline

The Pipeline executes a list of skills sequentially.

Usage

from yosrai import Pipeline

pipe = Pipeline(
    name="DataPipe",
    steps=[
        fetch_data_tool,
        process_data_agent,
        save_data_tool
    ]
)

result = pipe.run("start_id")

API Reference

yosrai.engine.agents.agent.Agent

Autonomous Actor in the Engine. Owns its identity (Instructions), its Hands (Tools), and its Voice (LLM). Runs a ReAct loop.

__init__(name, instructions, config=None, model=None, tools=[], api_key=None, callbacks=[], memory=None, response_model=None, event_manager=None, verbose=False, thinking=False, reflection=False, **kwargs)

run(input_data, response_model=None, stream=False, trace=False)

Run the agent loop (Sync).

Parameters:

Name	Type	Description	Default
input_data	Union[str, Context]	User query string or full Context object.	required
response_model	Any	Optional Pydantic model for structured output.	None
stream	bool	Whether to stream tokens to callbacks.	False
trace	bool	If True, return ExecutionResult with replay() capability.	False

Note

After calling run(), access usage via agent.last_usage:

result = agent.run("Hello") print(f"Cost: ${agent.last_usage.cost:.4f}")

arun(input_data, response_model=None, stream=False, trace=False) async

Run the agent loop (Async).

Parameters:

Name	Type	Description	Default
input_data	Union[str, Context]	User query string or full Context object.	required
response_model	Any	Optional Pydantic model for structured output.	None
trace	bool	If True, return ExecutionResult with replay() capability.	False

to_blueprint()

Serialize the Agent to a JSON-compatible blueprint.

Note: Tools are stored as their schemas only. When reconstructing, you must provide the actual tool functions via the tools parameter.

Returns:

Type	Description
Dict[str, Any]	Dict containing the agent's configuration.

from_blueprint(blueprint, tools=[], validate=True, **kwargs) classmethod

Reconstruct an Agent from a blueprint.

Parameters:

Name	Type	Description	Default
blueprint	Dict[str, Any]	The blueprint dict (from to_blueprint()).	required
tools	List[Callable]	List of actual tool functions to attach.	[]
validate	bool	Whether to validate the blueprint before reconstruction (default: True).	True
**kwargs	Any	Override any blueprint values (e.g., api_key).	{}

Returns:

Type	Description
Agent	A new Agent instance.

Raises:

Type	Description
BlueprintValidationError	If validation is enabled and the blueprint is invalid.

to_code()

Generate clean Python code that recreates this Agent.

Returns:

Type	Description
str	Python code string defining this agent as a complete, runnable file.

Note

Tools cannot be automatically serialized. If the agent uses tools, you must manually define them in the generated code.

yosrai.engine.agents.conductor.Conductor

Bases: Agent

A specialized Agent designed to orchestrate other Agents (Skills).

__init__(name, instructions, skills=[], shared_memory=None, planning=False, **kwargs)

run(input_data, response_model=None, stream=False, trace=False)

Run the Conductor.

If planning is enabled, it generates a plan first.

yosrai.engine.workflows.pipeline.Pipeline

Bases: Agent

A specialized Agent that executes a sequence of skills (Agents or functions) linearly. The output of step N becomes the input of step N+1.

__init__(name, steps, description='', **kwargs)

Initialize a Pipeline.

Parameters:

Name	Type	Description	Default
name	str	Name of the pipeline.	required
steps	List[Union[Agent, Callable]]	List of skills to execute in order.	required
description	str	Description of what this pipeline does (used as instructions).	''
**kwargs	Any	Agent config options.	{}

run(input_data, **kwargs)

Execute the pipeline steps.

as_skill(name=None)

Convert pipeline to a skill.

yosrai.engine.agents.skill_config.SkillConfig

Bases: BaseModel

Configuration for a Conductor skill.

to_blueprint()

Serialize skill config.

wrap()

Wrap the skill with retry/timeout logic.