Skip to content
BeeAI Framework

Agents

Overview

Section titled “Overview”

An AI Agent is a system built on language models (LLMs or SLMs) that can solve complex tasks through structured reasoning and autonomous or human-assisted actions. The BeeAI Framework serves as the orchestration layer that enables agents to do this and more:

  • Coordinate with LLMs: Manages communication between your agent and language models
  • Tool Management: Provides agents with access to external tools and handles their execution
  • Response Processing: Processes and validates tool outputs and model responses
  • Memory Management: Maintains conversation context and state across interactions
  • Error Handling: Manages retries, timeouts, and graceful failure recovery
  • Event Orchestration: Emits detailed events for monitoring and debugging agent behavior

Unlike basic chatbots, agents built with this framework can perform multi-step reasoning, use tools to interact with external systems, maintain context across interactions, and adapt based on feedback. These capabilities make them ideal for planning, research, analysis, and complex execution tasks.

Customizing Agent Behavior

Section titled “Customizing Agent Behavior”

You can customize your agent’s behavior in several key ways:

1. Configuring the Language Model Backend

Section titled “1. Configuring the Language Model Backend”

The backend system manages your connection to different language model providers. The BeeAI Framework supports multiple LLM providers through a unified interface. Learn more about available backends and how to set their parameters in our backend documentation.

from beeai_framework.backend import ChatModel
from beeai_framework.agents.requirement import RequirementAgent


# Using Ollama (local models)
llm = ChatModel.from_name("ollama:granite3.3")


# Using OpenAI
llm = ChatModel.from_name("openai:gpt-5-mini")


# Using Anthropic
llm = ChatModel.from_name("anthropic:claude-sonnet-4")


agent = RequirementAgent(
    llm=llm,
    # ... other configuration
)

Backend Features:

  • Unified Interface: Work with different providers using the same API
  • Model Parameters: Configure temperature, max tokens, and other model settings
  • Provider Support: OpenAI, Anthropic, Ollama, Groq, and more
  • Local & Cloud: Support for both local and cloud-hosted models

2. Setting the System Prompt

Section titled “2. Setting the System Prompt”

The system prompt defines your agent’s behavior, personality, and capabilities. You can configure this through several parameters when initializing an agent:

agent = RequirementAgent(
    llm="ollama:granite4:micro",
    role="You are a helpful research assistant specializing in academic papers",
    instructions=[
        "Always provide citations for your sources",
        "Focus on peer-reviewed research when possible",
        "Explain complex concepts in simple terms"
    ],
    notes=[
        "Be especially careful about medical or legal advice",
        "If unsure about a fact, acknowledge the uncertainty"
    ],
    name="Research Assistant",
    description="An AI agent that helps with academic research tasks"
)

Prompt Parameters:

  • role: Defines the agent’s persona and primary function
  • instructions: List of specific behavioral guidelines
  • notes: Additional context or special considerations
  • name and description: Help identify the agent’s purpose and are helpful when using the HandoffTool, Serve module, and when you need to access agent metadata via agent.meta

Setting instructions, notes, role will be integrated into the framework provided system prompt template. If you want to completely override the framework provided system prompt template, you can provide a custom prompt template.

3. Configuring Agent Run Options

Section titled “3. Configuring Agent Run Options”

When executing an agent, you can provide additional options to guide its behavior and execution settings:

Setting Execution Settings and Guiding Agent Run Behavior

Section titled “Setting Execution Settings and Guiding Agent Run Behavior”
response = await agent.run(
    Analyze the latest AI research trends",
    expected_output="A structured summary with key findings and recommendations",
  # expected_output can also receive a Pydantic Model or a JSON Schema
    backstory="The user is preparing for a conference presentation on AI trends",
    total_max_retries=5,
    max_retries_per_step=2,
    max_iterations=15,
)


#structured outputs can be accessed like this:
print(response.output_structured)

Available Options:

  • expected_output: Guides the agent toward a specific unstructured or structured output format. output_structured is defined only when expected_output is a Pydantic model or a JSON schema. However, the text representation is always available via response.output.
  • backstory: Provides additional context to help the agent understand the user’s situation
  • total_max_retries: Controls the total number of retry attempts across the entire agent execution
  • max_retries_per_step: Limits retries for individual steps (like tool calls or model responses)
  • max_iterations: Sets the maximum number of reasoning cycles the agent can perform

4. Adding Tools

Section titled “4. Adding Tools”

Enhance your agent’s capabilities by providing it with tools to interact with external systems. Learn more about beeai provided tools and creating custom tools in our tools documentation.

from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.weather import OpenMeteoTool


agent = RequirementAgent(
    llm="ollama:granite4",
    tools=[
        DuckDuckGoSearchTool(),
        OpenMeteoTool(),
        # Add more tools as needed
    ]
)

5. Configuring Memory

Section titled “5. Configuring Memory”

Memory allows your agent to maintain context across multiple interactions. Different memory types serve different use cases. Learn more about our built in options in the memory documentation.

from beeai_framework.memory import TokenMemory, UnconstrainedMemory
from beeai_framework.agents.requirement import RequirementAgent


agent = RequirementAgent(
    llm="ollama:granite4:micro",
    memory=TokenMemory(max_tokens=20*1024)
)

Additional Agent Options

Section titled “Additional Agent Options”

Agent Types

Section titled “Agent Types”

BeeAI Framework provides several agent implementations:

Requirement Agent

Section titled “Requirement Agent”

This agent provides the reliability needed for production scenarios through a rule system that defines execution constraints while keeping problem-solving flexibility intact. Unlike traditional approaches that require complex orchestration code, RequirementAgent uses a declarative interface where you define requirements and let the framework enforce them automatically.

Learn more about RequirementAgent in its dedicated page or in the blog post.

from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.agents.requirement.requirements.conditional import ConditionalRequirement
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.think import ThinkTool
from beeai_framework.tools.weather import OpenMeteoTool
from beeai_framework.backend import ChatModel
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware


agent = RequirementAgent(
    llm=ChatModel.from_name("ollama:granite4:micro"),
    tools=[
        ThinkTool(),             # to reason
        OpenMeteoTool(),         # retrieve weather data
        DuckDuckGoSearchTool()   # search web
    ],
    instructions="Plan activities for a given destination based on current weather and events.",
    requirements=[
        # Force thinking first
        ConditionalRequirement(ThinkTool, force_at_step=1),
        # Search only after getting weather and at least once
        ConditionalRequirement(DuckDuckGoSearchTool, only_after=[OpenMeteoTool], min_invocations=1),
        # Weather tool be used at least once but not consecutively
        ConditionalRequirement(OpenMeteoTool, consecutive_allowed=False, min_invocations=1),
    ]
)


# Run with execution logging
response = await agent.run("What to do in Boston?").middleware(GlobalTrajectoryMiddleware())
print(f"Final Answer: {response.answer.text}")

Lite Agent

Section titled “Lite Agent”

An agent that leverages a language model and a suite of tools to solve problems. The agent does not have any system prompt.

This design is ideal for:

  • Exploring the raw capabilities of a language model without bias from a framework’s built‑in prompt
  • Developers who want to understand how to build an agent from scratch
import asyncio


from beeai_framework.agents.lite import LiteAgent
from beeai_framework.backend import ChatModel, ChatModelOutput, ChatModelParameters, SystemMessage
from beeai_framework.emitter import EventMeta
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.think import ThinkTool
from beeai_framework.tools.weather import OpenMeteoTool




async def main() -> None:
    agent = LiteAgent(
        llm=ChatModel.from_name("ollama:granite4:micro", ChatModelParameters(stream=True)),
        tools=[ThinkTool(), OpenMeteoTool(), DuckDuckGoSearchTool()],
        middlewares=[GlobalTrajectoryMiddleware()],
    )


    # Optionally set a custom system prompt
    await agent.memory.add(SystemMessage("You are a helpful assistant."))


    @agent.emitter.on("final_answer")
    def stream_final_answer(data: ChatModelOutput, meta: EventMeta) -> None:
        print(data.get_text_content())  # emits chunks


    await agent.run("Hello")




if __name__ == "__main__":
    asyncio.run(main())

ReAct Agent

Section titled “ReAct Agent”

The ReActAgent implements the ReAct (Reasoning and Acting) pattern, which structures agent behavior into a cyclical process of reasoning, action, and observation.

This pattern allows agents to reason about a task, take actions using tools, observe results, and continue reasoning until reaching a conclusion.

Let’s see how a ReActAgent approaches a simple question:

Input prompt: “What is the current weather in Las Vegas?”

First iteration:

thought: I need to retrieve the current weather in Las Vegas. I can use the OpenMeteo function to get the current weather forecast for a location.
tool_name: OpenMeteo
tool_input: {"location": {"name": "Las Vegas"}, "start_date": "2024-10-17", "end_date": "2024-10-17", "temperature_unit": "celsius"}

Second iteration:

thought: I have the current weather in Las Vegas in Celsius.
final_answer: The current weather in Las Vegas is 20.5°C with an apparent temperature of 18.3°C.
import asyncio
import logging
import os
import sys
import tempfile
import traceback
from typing import Any


from dotenv import load_dotenv


from beeai_framework.agents.react import ReActAgent
from beeai_framework.backend import ChatModel, ChatModelParameters
from beeai_framework.emitter import EmitterOptions, EventMeta
from beeai_framework.errors import FrameworkError
from beeai_framework.logger import Logger
from beeai_framework.memory import TokenMemory
from beeai_framework.tools import AnyTool
from beeai_framework.tools.code import LocalPythonStorage, PythonTool
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.search.wikipedia import WikipediaTool
from beeai_framework.tools.weather import OpenMeteoTool
from examples.helpers.io import ConsoleReader


# Load environment variables
load_dotenv()


# Configure logging - using DEBUG instead of trace
logger = Logger("app", level=logging.DEBUG)


reader = ConsoleReader()




def create_agent() -> ReActAgent:
    """Create and configure the agent with tools and LLM"""


    # Other models to try:
    # "llama3.1"
    # "granite4.1:8b"
    # "deepseek-r1"
    # ensure the model is pulled before running
    llm = ChatModel.from_name(
        "ollama:granite4:micro",
        ChatModelParameters(temperature=0),
    )


    # Configure tools
    tools: list[AnyTool] = [
        WikipediaTool(),
        OpenMeteoTool(),
        DuckDuckGoSearchTool(),
    ]


    # Add code interpreter tool if URL is configured
    code_interpreter_url = os.getenv("CODE_INTERPRETER_URL")
    if code_interpreter_url:
        tools.append(
            PythonTool(
                code_interpreter_url,
                LocalPythonStorage(
                    local_working_dir=tempfile.mkdtemp("code_interpreter_source"),
                    interpreter_working_dir=os.getenv("CODE_INTERPRETER_TMPDIR", "./tmp/code_interpreter_target"),
                ),
            )
        )


    # Create agent with memory and tools
    agent = ReActAgent(llm=llm, tools=tools, memory=TokenMemory(llm))


    return agent




def process_agent_events(data: Any, event: EventMeta) -> None:
    """Process agent events and log appropriately"""


    if event.name == "error":
        reader.write("Agent 🤖 : ", FrameworkError.ensure(data.error).explain())
    elif event.name == "retry":
        reader.write("Agent 🤖 : ", "retrying the action...")
    elif event.name == "update":
        reader.write(f"Agent({data.update.key}) 🤖 : ", data.update.parsed_value)
    elif event.name == "start":
        reader.write("Agent 🤖 : ", "starting new iteration")
    elif event.name == "success":
        reader.write("Agent 🤖 : ", "success")




async def main() -> None:
    """Main application loop"""


    # Create agent
    agent = create_agent()


    # Log code interpreter status if configured
    code_interpreter_url = os.getenv("CODE_INTERPRETER_URL")
    if code_interpreter_url:
        reader.write(
            "🛠️ System: ",
            f"The code interpreter tool is enabled. Please ensure that it is running on {code_interpreter_url}",
        )


    reader.write("🛠️ System: ", "Agent initialized with Wikipedia, DuckDuckGo, and Weather tools.")


    # Main interaction loop with user input
    for prompt in reader:
        # Run agent with the prompt
        response = await agent.run(
            prompt,
            max_retries_per_step=3,
            total_max_retries=10,
            max_iterations=20,
        ).on("*", process_agent_events, EmitterOptions(match_nested=False))


        reader.write("Agent 🤖 : ", response.last_message.text)




if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

Tool Calling Agent

Section titled “Tool Calling Agent”

The ToolCallingAgent is optimized for scenarios where tool usage is the primary focus. It handles tool calls more efficiently and can execute multiple tools in parallel.

import asyncio
import logging
import sys
import traceback
from typing import Any


from dotenv import load_dotenv


from beeai_framework.agents.tool_calling import ToolCallingAgent
from beeai_framework.backend import ChatModel
from beeai_framework.emitter import EventMeta
from beeai_framework.errors import FrameworkError
from beeai_framework.logger import Logger
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.tools.weather import OpenMeteoTool
from examples.helpers.io import ConsoleReader


# Load environment variables
load_dotenv()


# Configure logging - using DEBUG instead of trace
logger = Logger("app", level=logging.DEBUG)


reader = ConsoleReader()




def process_agent_events(data: Any, event: EventMeta) -> None:
    """Process agent events and log appropriately"""


    if event.name == "start":
        reader.write("Agent (debug) 🤖 : ", "starting new iteration")
    elif event.name == "success":
        reader.write("Agent (debug) 🤖 : ", data.state.memory.messages[-1])




async def main() -> None:
    """Main application loop"""


    # Create agent
    agent = ToolCallingAgent(
        llm=ChatModel.from_name("ollama:llama3.1"), memory=UnconstrainedMemory(), tools=[OpenMeteoTool()]
    )


    # Main interaction loop with user input
    for prompt in reader:
        response = await agent.run(prompt).on("*", process_agent_events)
        reader.write("Agent 🤖 : ", response.last_message.text)


    print("======DONE (showing the full message history)=======")


    # pyrefly: ignore [unbound-name]
    messages = response.state.memory.messages
    for msg in messages:
        print(msg)




if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

Custom Agent

Section titled “Custom Agent”

For advanced use cases, you can create your own agent implementation by extending the BaseAgent class.

import asyncio
import sys
import traceback
from typing import Unpack


from pydantic import BaseModel, Field


from beeai_framework.adapters.ollama import OllamaChatModel
from beeai_framework.agents import AgentMeta, AgentOptions, AgentOutput, BaseAgent
from beeai_framework.backend import AnyMessage, AssistantMessage, ChatModel, SystemMessage, UserMessage
from beeai_framework.context import RunContext
from beeai_framework.emitter import Emitter
from beeai_framework.errors import FrameworkError
from beeai_framework.memory import BaseMemory, UnconstrainedMemory
from beeai_framework.runnable import runnable_entry




class State(BaseModel):
    thought: str
    final_answer: str




class CustomAgent(BaseAgent):
    def __init__(self, llm: ChatModel, memory: BaseMemory) -> None:
        super().__init__()
        self.model = llm
        self._memory = memory


    @property
    def memory(self) -> BaseMemory:
        return self._memory


    @memory.setter
    def memory(self, memory: BaseMemory) -> None:
        self._memory = memory


    def _create_emitter(self) -> Emitter:
        return Emitter.root().child(
            namespace=["agent", "custom"],
            creator=self,
        )


    @runnable_entry
    async def run(self, input: str | list[AnyMessage], /, **kwargs: Unpack[AgentOptions]) -> AgentOutput:
        async def handler(context: RunContext) -> AgentOutput:
            class CustomSchema(BaseModel):
                thought: str = Field(description="Describe your thought process before coming with a final answer")
                final_answer: str = Field(
                    description="Here you should provide concise answer to the original question."
                )


            response = await self.model.run(
                [
                    SystemMessage("You are a helpful assistant. Always use JSON format for your responses."),
                    *(self.memory.messages if self.memory is not None else []),
                    *([UserMessage(input)] if isinstance(input, str) else input),
                ],
                response_format=CustomSchema,
                max_retries=kwargs.get("total_max_retries", 3),
                signal=context.signal,
            )
            assert isinstance(response.output_structured, CustomSchema)


            result = AssistantMessage(response.output_structured.final_answer)
            await self.memory.add(result) if self.memory else None


            return AgentOutput(
                output=[result],
                context={
                    "state": State(
                        thought=response.output_structured.thought,
                        final_answer=response.output_structured.final_answer,
                    )
                },
            )


        return await handler(RunContext.get())


    @property
    def meta(self) -> AgentMeta:
        return AgentMeta(
            name="CustomAgent",
            description="Custom Agent is a simple LLM agent.",
            tools=[],
        )




async def main() -> None:
    agent = CustomAgent(
        llm=OllamaChatModel("granite3.3"),
        memory=UnconstrainedMemory(),
    )


    response = await agent.run([UserMessage("Why is the sky blue?")])
    print(response.context.get("state"))




if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

Multi-Agent Hand-offs

Section titled “Multi-Agent Hand-offs”

Create a team of specialized agents that can collaborate:

import asyncio


from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.agents.requirement.requirements.conditional import ConditionalRequirement
from beeai_framework.backend import ChatModel
from beeai_framework.errors import FrameworkError
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools import Tool
from beeai_framework.tools.handoff import HandoffTool
from beeai_framework.tools.search.wikipedia import WikipediaTool
from beeai_framework.tools.think import ThinkTool
from beeai_framework.tools.weather import OpenMeteoTool




async def main() -> None:
    knowledge_agent = RequirementAgent(
        llm=ChatModel.from_name("ollama:granite4:micro"),
        tools=[ThinkTool(), WikipediaTool()],
        requirements=[ConditionalRequirement(ThinkTool, force_at_step=1)],
        role="Knowledge Specialist",
        instructions="Provide answers to general questions about the world.",
    )


    weather_agent = RequirementAgent(
        llm=ChatModel.from_name("ollama:granite4:micro"),
        tools=[OpenMeteoTool()],
        role="Weather Specialist",
        instructions="Provide weather forecast for a given destination.",
    )


    main_agent = RequirementAgent(
        name="MainAgent",
        llm=ChatModel.from_name("ollama:granite4:micro"),
        tools=[
            ThinkTool(),
            HandoffTool(
                knowledge_agent,
                name="KnowledgeLookup",
                description="Consult the Knowledge Agent for general questions.",
            ),
            HandoffTool(
                weather_agent,
                name="WeatherLookup",
                description="Consult the Weather Agent for forecasts.",
            ),
        ],
        requirements=[ConditionalRequirement(ThinkTool, force_at_step=1)],
        # Log all tool calls to the console for easier debugging
        middlewares=[GlobalTrajectoryMiddleware(included=[Tool])],
    )


    question = "If I travel to Rome next weekend, what should I expect in terms of weather, and also tell me one famous historical landmark there?"
    print(f"User: {question}")


    try:
        response = await main_agent.run(question, expected_output="Helpful and clear response.")
        print("Agent:", response.last_message.text)
    except FrameworkError as err:
        print("Error:", err.explain())




if __name__ == "__main__":
    asyncio.run(main())

Workflows

Section titled “Workflows”

For complex applications, you can create multi-agent workflows where specialized agents collaborate.

import asyncio
import sys
import traceback


from beeai_framework.backend import ChatModel
from beeai_framework.emitter import EmitterOptions
from beeai_framework.errors import FrameworkError
from beeai_framework.tools.search.wikipedia import WikipediaTool
from beeai_framework.tools.weather import OpenMeteoTool
from beeai_framework.workflows.agent import AgentWorkflow, AgentWorkflowInput
from examples.helpers.io import ConsoleReader




async def main() -> None:
    llm = ChatModel.from_name("ollama:llama3.1")
    workflow = AgentWorkflow(name="Smart assistant")


    workflow.add_agent(
        name="Researcher",
        role="A diligent researcher.",
        instructions="You look up and provide information about a specific topic.",
        tools=[WikipediaTool()],
        llm=llm,
    )


    workflow.add_agent(
        name="WeatherForecaster",
        role="A weather reporter.",
        instructions="You provide detailed weather reports.",
        tools=[OpenMeteoTool()],
        llm=llm,
    )


    workflow.add_agent(
        name="DataSynthesizer",
        role="A meticulous and creative data synthesizer",
        instructions="You can combine disparate information into a final coherent summary.",
        llm=llm,
    )


    reader = ConsoleReader()


    reader.write("Assistant 🤖 : ", "What location do you want to learn about?")
    for prompt in reader:
        await (
            workflow.run(
                inputs=[
                    AgentWorkflowInput(prompt="Provide a short history of the location.", context=prompt),
                    AgentWorkflowInput(
                        prompt="Provide a comprehensive weather summary for the location today.",
                        expected_output="Essential weather details such as chance of rain, temperature and wind. Only report information that is available.",
                    ),
                    AgentWorkflowInput(
                        prompt="Summarize the historical and weather data for the location.",
                        expected_output="A paragraph that describes the history of the location, followed by the current weather conditions.",
                    ),
                ]
            )
            .on(
                # Event Matcher -> match agent's 'success' events
                lambda event: isinstance(event.creator, ChatModel) and event.name == "success",
                # log data to the console
                lambda data, event: reader.write(
                    "->Got response from the LLM",
                    "  \n->".join([str(message.content[0].model_dump()) for message in data.value.messages]),
                ),
                EmitterOptions(match_nested=True),
            )
            .on(
                "success",
                lambda data, event: reader.write(
                    f"->Step '{data.step}' has been completed with the following outcome."
                    f"\n\n{data.state.final_answer}\n\n",
                    data.model_dump(exclude={"data"}),
                ),
            )
        )
        reader.write("Assistant 🤖 : ", "What location do you want to learn about?")




if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

Examples

Section titled “Examples”

Python

Explore reference agent implementations in Python

TypeScript

Explore reference agent implementations in TypeScript