Agents

Core agent classes for building AI applications. All agents share the same public API: chat(), chat_with_history(), chat_stream(), and on_event().

from sagewai import UniversalAgent

agent = UniversalAgent(name="helper", model="gpt-4o")
response = await agent.chat("Hello!")

BaseAgent

Abstract foundation for all agents. Provides the agentic tool-calling loop, guardrails, context compaction, event system, and budget enforcement. You never instantiate BaseAgent directly.

from sagewai import BaseAgent

Constructor

BaseAgent(
    name: str,
    model: str = "gpt-4o",
    system_prompt: str = "",
    tools: list[ToolSpec] | None = None,
    temperature: float = 0.7,
    max_tokens: int | None = None,
    max_iterations: int = 10,
    strategy: ExecutionStrategy | None = None,
    memory: Any = None,
    guardrails: list[Guardrail] | None = None,
    max_context_tokens: int | None = None,
    rate_limiter: RateLimiter | None = None,
    api_base: str | None = None,
    api_key: str | None = None,
    custom_llm_provider: str | None = None,
    directives: Any = None,
)
ParameterTypeDefaultDescription
namestrrequiredAgent identifier used in logs, events, and admin
modelstr"gpt-4o"LLM model string (any LiteLLM-supported model)
system_promptstr""System message prepended to all conversations
toolslist[ToolSpec] | NoneNoneTools the agent can invoke
temperaturefloat0.7LLM sampling temperature
max_tokensint | NoneNoneMax output tokens per LLM call
max_iterationsint10Max tool-calling loop iterations
strategyExecutionStrategy | NoneNoneExecution strategy (defaults to ReActStrategy)
memoryAnyNoneMemory provider for RAG (e.g. ContextEngine)
guardrailslist[Guardrail] | NoneNoneInput/output validation guardrails
max_context_tokensint | NoneNoneEnables automatic context compaction
rate_limiterRateLimiter | NoneNonePer-agent rate limiting
api_basestr | NoneNoneOverride the LLM API base URL
api_keystr | NoneNoneOverride the LLM API key
custom_llm_providerstr | NoneNoneForce a specific LiteLLM provider
directivesAnyNoneEnable directive preprocessing (True, False, or DirectiveEngine instance)

Methods

MethodSignatureReturnsDescription
chatasync chat(message: str)strSend a message, get text response
chat_with_historyasync chat_with_history(messages: list[ChatMessage])ChatMessageRun with explicit conversation history
chat_streamasync chat_stream(message: str)AsyncGenerator[str, None]Stream text chunks
on_eventon_event(callback: EventCallback)NoneRegister event listener

UniversalAgent

LiteLLM-based agent supporting 100+ LLM providers: OpenAI, Anthropic, Google, Mistral, Cohere, Azure, and more.

from sagewai import UniversalAgent

agent = UniversalAgent(
    name="analyst",
    model="claude-sonnet-4-20250514",
    system_prompt="You are a data analyst.",
    temperature=0.3,
)
response = await agent.chat("Summarize Q4 revenue trends.")

Inherits all BaseAgent constructor parameters. Same public API.


GoogleNativeAgent

Native Gemini agent using the google-genai SDK. Direct access to Gemini features without the LiteLLM proxy layer.

from sagewai import GoogleNativeAgent

agent = GoogleNativeAgent(
    name="gemini",
    model="gemini-2.0-flash",
    system_prompt="You are a helpful assistant.",
)
response = await agent.chat("Explain quantum entanglement.")

Inherits all BaseAgent constructor parameters. Same public API.


SequentialAgent

Execute sub-agents in order, passing each output as input to the next.

from sagewai import SequentialAgent, UniversalAgent

researcher = UniversalAgent(name="researcher", model="gpt-4o")
writer = UniversalAgent(name="writer", model="gpt-4o")
pipeline = SequentialAgent(name="pipeline", agents=[researcher, writer])
result = await pipeline.chat("Write about quantum computing")

Constructor

SequentialAgent(
    name: str,
    agents: list[BaseAgent],
    *,
    durability: DurabilityMode = DurabilityMode.NONE,
    workflow_store: WorkflowStore | None = None,
    **kwargs,
)
ParameterTypeDefaultDescription
namestrrequiredPipeline name
agentslist[BaseAgent]requiredOrdered list of sub-agents
durabilityDurabilityModeNONENONE or CHECKPOINT for durable execution
workflow_storeWorkflowStore | NoneNoneStore for checkpoint persistence

Additional method: async resume(run_id: str) -> str -- resume a checkpointed run.


ParallelAgent

Execute sub-agents concurrently and merge their outputs.

from sagewai import ParallelAgent, UniversalAgent

analyst = UniversalAgent(name="analyst", model="gpt-4o")
critic = UniversalAgent(name="critic", model="gpt-4o")
panel = ParallelAgent(name="panel", agents=[analyst, critic])
result = await panel.chat("Evaluate this proposal")

Constructor

ParallelAgent(
    name: str,
    agents: list[BaseAgent],
    merge: Callable[[list[str]], str] | None = None,
    *,
    durability: DurabilityMode = DurabilityMode.NONE,
    workflow_store: WorkflowStore | None = None,
    **kwargs,
)
ParameterTypeDefaultDescription
namestrrequiredAgent name
agentslist[BaseAgent]requiredSub-agents to run concurrently
mergeCallable[[list[str]], str] | NoneNoneCustom merge function (default: join with newlines)

LoopAgent

Repeat a sub-agent until a condition is met or max iterations reached.

from sagewai import LoopAgent, UniversalAgent

refiner = UniversalAgent(name="refiner", model="gpt-4o")
loop = LoopAgent(
    name="refinement-loop",
    agent=refiner,
    max_iterations=3,
    should_stop=lambda result, i: "DONE" in result,
)
result = await loop.chat("Improve this text iteratively")

Constructor

LoopAgent(
    name: str,
    agent: BaseAgent,
    should_stop: Callable[[str, int], bool] | None = None,
    *,
    durability: DurabilityMode = DurabilityMode.NONE,
    workflow_store: WorkflowStore | None = None,
    **kwargs,
)
ParameterTypeDefaultDescription
namestrrequiredAgent name
agentBaseAgentrequiredThe sub-agent to loop
should_stopCallable[[str, int], bool] | NoneNoneStop condition receiving (result, iteration)
max_iterationsint10Maximum loop iterations (inherited from BaseAgent)

ConditionalAgent

Route input to different agents based on a sync or async condition function.

from sagewai import ConditionalAgent, UniversalAgent

greeter = UniversalAgent(name="greeter", model="gpt-4o-mini")
researcher = UniversalAgent(name="researcher", model="gpt-4o")

def route(message: str) -> str:
    return "greeter" if "hello" in message.lower() else "researcher"

router = ConditionalAgent(
    name="router",
    condition=route,
    agents={"greeter": greeter, "researcher": researcher},
)
result = await router.chat("Hello there!")

agent_as_tool

Wrap an agent so another agent can invoke it as a tool.

from sagewai import agent_as_tool, UniversalAgent

researcher = UniversalAgent(name="researcher", model="gpt-4o")
tool = agent_as_tool(researcher, description="Research any topic")

orchestrator = UniversalAgent(
    name="orchestrator",
    model="gpt-4o",
    tools=[tool],
)

Signature

agent_as_tool(
    agent: BaseAgent,
    description: str,
    *,
    tool_name: str | None = None,
) -> ToolSpec
ParameterTypeDefaultDescription
agentBaseAgentrequiredThe sub-agent to wrap
descriptionstrrequiredDescription shown to the LLM
tool_namestr | NoneNoneOverride auto-generated tool name

AgentConfig

Configuration object stored on every agent instance at agent.config.

from sagewai import AgentConfig

config = AgentConfig(
    name="my-agent",
    model="gpt-4o",
    system_prompt="You are helpful.",
)

Key Fields

FieldTypeDefaultDescription
namestrrequiredUnique agent identifier
modelstr"gpt-4o"LLM model string
system_promptstr""System prompt
toolslist[ToolSpec][]Available tools
inferenceInferenceParamsInferenceParams()LLM inference parameters
max_iterationsint10Max tool-calling iterations
strategyAnyNoneExecution strategy
memoryAnyNoneMemory provider
auto_learnboolFalseAuto-extract facts from conversations
learn_every_n_turnsint5Auto-learn frequency
durabilitystr"none"Durability mode ("none" or "checkpoint")
project_idstr | NoneNoneProject scope for multi-tenancy
directivesAnyNoneDirective engine configuration

InferenceParams

LLM inference parameters with validation and preset support.

from sagewai import InferenceParams

# Custom
params = InferenceParams(temperature=0.3, top_p=0.8, max_tokens=4096)

# From preset
from sagewai.models.inference import InferencePreset
params = InferenceParams.from_preset(InferencePreset.CREATIVE)

Fields

FieldTypeDefaultDescription
temperaturefloat0.7Sampling temperature (0.0-2.0)
top_pfloat | NoneNoneNucleus sampling threshold (0.0-1.0)
top_kint | NoneNoneTop-k sampling
max_tokensint | NoneNoneMax output tokens
frequency_penaltyfloat | NoneNoneFrequency penalty (-2.0 to 2.0)
presence_penaltyfloat | NoneNonePresence penalty (-2.0 to 2.0)
stop_sequenceslist[str] | NoneNoneStop sequences
api_basestr | NoneNoneCustom API base URL
api_keystr | NoneNoneCustom API key
custom_llm_providerstr | NoneNoneForce LiteLLM provider
timeoutfloat | None120.0LLM call timeout in seconds
fallback_modelslist[str][]Ordered fallback models on failure

Presets

PresetTemperatureTop P
PRECISE0.10.9
BALANCED0.50.95
CREATIVE0.91.0
EXPERIMENTAL1.51.0