Overview
An AI agent is a system built on large language models (LLMs) that can solve complex tasks through structured reasoning and autonomous actions. Unlike basic chatbots, agents can:
- Perform multi-step reasoning
- Use tools to interact with external systems
- Maintain context across interactions
- Adapt based on feedback
These capabilities make them ideal for planning, research, analysis, and complex execution.
Dive deeper into the concepts behind AI agents in this research article from IBM.
Supported in Python and TypeScript.
Agent Types
BeeAI Framework provides several agent implementations for different use cases.
Check out our new experimental RequirementAgent that combines the power of LLMs, tools, and requirements, all wrapped in a declarative interface.
This approach will soon become a default building block for building agents and will replace the others.
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.
During execution, the agent emits partial updates as it generates each line, followed by complete updates. Updates follow a strict order: first all partial updates for “thought,” then a complete “thought” update, then moving to the next component.
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 import AgentExecutionConfig
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"
# "granite3.3:8b"
# "deepseek-r1"
# ensure the model is pulled before running
llm = ChatModel.from_name(
"ollama:granite3.3:8b",
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=prompt,
execution=AgentExecutionConfig(max_retries_per_step=3, total_max_retries=10, max_iterations=20),
).on("*", process_agent_events, EmitterOptions(match_nested=False))
reader.write("Agent 🤖 : ", response.result.text)
if __name__ == "__main__":
try:
asyncio.run(main())
except FrameworkError as e:
traceback.print_exc()
sys.exit(e.explain())
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.result.text)
print("======DONE (showing the full message history)=======")
for msg in response.memory.messages:
print(msg)
if __name__ == "__main__":
try:
asyncio.run(main())
except FrameworkError as e:
traceback.print_exc()
sys.exit(e.explain())
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 pydantic import BaseModel, Field, InstanceOf
from beeai_framework.adapters.ollama import OllamaChatModel
from beeai_framework.agents import AgentMeta, BaseAgent, BaseAgentRunOptions
from beeai_framework.backend import AnyMessage, AssistantMessage, ChatModel, SystemMessage, UserMessage
from beeai_framework.context import Run, RunContext
from beeai_framework.emitter import Emitter
from beeai_framework.errors import FrameworkError
from beeai_framework.memory import BaseMemory, UnconstrainedMemory
class State(BaseModel):
thought: str
final_answer: str
class RunInput(BaseModel):
message: InstanceOf[AnyMessage]
class CustomAgentRunOptions(BaseAgentRunOptions):
max_retries: int | None = None
class CustomAgentRunOutput(BaseModel):
message: InstanceOf[AnyMessage]
state: State
class CustomAgent(BaseAgent[CustomAgentRunOutput]):
memory: BaseMemory | None = None
def __init__(self, llm: ChatModel, memory: BaseMemory) -> None:
super().__init__()
self.model = llm
self.memory = memory
def _create_emitter(self) -> Emitter:
return Emitter.root().child(
namespace=["agent", "custom"],
creator=self,
)
def run(
self,
run_input: RunInput,
options: CustomAgentRunOptions | None = None,
) -> Run[CustomAgentRunOutput]:
async def handler(context: RunContext) -> CustomAgentRunOutput:
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.create_structure(
schema=CustomSchema,
messages=[
SystemMessage("You are a helpful assistant. Always use JSON format for your responses."),
*(self.memory.messages if self.memory is not None else []),
run_input.message,
],
max_retries=options.max_retries if options else None,
abort_signal=context.signal,
)
result = AssistantMessage(response.object["final_answer"])
await self.memory.add(result) if self.memory else None
return CustomAgentRunOutput(
message=result,
state=State(thought=response.object["thought"], final_answer=response.object["final_answer"]),
)
return self._to_run(
handler, signal=options.signal if options else None, run_params={"input": run_input, "options": options}
)
@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:8b"),
memory=UnconstrainedMemory(),
)
response = await agent.run(RunInput(message=UserMessage("Why is the sky blue?")))
print(response.state)
if __name__ == "__main__":
try:
asyncio.run(main())
except FrameworkError as e:
traceback.print_exc()
sys.exit(e.explain())
Customizing Agent Behavior
You can customize your agent’s behavior in five ways:
1. Setting Execution Policy
Control how the agent runs by configuring retries, timeouts, and iteration limits.
response = await agent.run(
prompt=prompt,
execution=AgentExecutionConfig(max_retries_per_step=3, total_max_retries=10, max_iterations=20),
).on("*", process_agent_events, EmitterOptions(match_nested=False))
The default is zero retries and no timeout. For complex tasks, increasing the max_iterations is recommended.
2. Overriding Prompt Templates
Customize how the agent formats prompts, including the system prompt that defines its behavior.
The agent uses several templates that you can override:
- System Prompt - Defines the agent’s behavior and capabilities
- User Prompt - Formats the user’s input
- Tool Error - Handles tool execution errors
- Tool Input Error - Handles validation errors
- Tool No Result Error - Handles empty results
- Tool Not Found Error - Handles references to missing tools
- Invalid Schema Error - Handles parsing errors
import sys
import traceback
from beeai_framework.agents.react.runners.default.prompts import (
SystemPromptTemplate,
ToolDefinition,
)
from beeai_framework.errors import FrameworkError
from beeai_framework.tools.weather import OpenMeteoTool
from beeai_framework.utils.strings import to_json
def main() -> None:
tool = OpenMeteoTool()
tool_def = ToolDefinition(
name=tool.name,
description=tool.description,
input_schema=to_json(tool.input_schema.model_json_schema()),
)
# Render the granite system prompt
prompt = SystemPromptTemplate.render(
instructions="You are a helpful AI assistant!", tools=[tool_def], tools_length=1
)
print(prompt)
if __name__ == "__main__":
try:
main()
except FrameworkError as e:
traceback.print_exc()
sys.exit(e.explain())
Enhance your agent’s capabilities by providing it with tools to interact with external systems.
agent = ReActAgent(
llm=llm,
tools=[DuckDuckGoSearchTool(), OpenMeteoTool()],
memory=UnconstrainedMemory()
)
Available tools include:
- Search tools (
DuckDuckGoSearchTool)
- Weather tools (
OpenMeteoTool)
- Knowledge tools (
LangChainWikipediaTool)
- And many more in the
beeai_framework.tools module
4. Configuring Memory
Memory allows your agent to maintain context across multiple interactions.
Several memory types are available for different use cases:
- UnconstrainedMemory - For unlimited storage
- SlidingMemory - For keeping only the most recent messages
- TokenMemory - For managing token limits
- SummarizeMemory - For summarizing previous conversations
agent = ReActAgent(
llm=llm,
tools=[DuckDuckGoSearchTool(), OpenMeteoTool()],
memory=UnconstrainedMemory()
)
import asyncio
import sys
import traceback
from beeai_framework.agents import AgentExecutionConfig
from beeai_framework.agents.react import ReActAgent
from beeai_framework.backend import AssistantMessage, ChatModel, UserMessage
from beeai_framework.errors import FrameworkError
from beeai_framework.memory import UnconstrainedMemory
# Initialize the memory and LLM
memory = UnconstrainedMemory()
def create_agent() -> ReActAgent:
llm = ChatModel.from_name("ollama:granite3.3:8b")
# Initialize the agent
agent = ReActAgent(llm=llm, memory=memory, tools=[])
return agent
async def main() -> None:
# Create user message
user_input = "Hello world!"
user_message = UserMessage(user_input)
# Await adding user message to memory
await memory.add(user_message)
print("Added user message to memory")
# Create agent
agent = create_agent()
response = await agent.run(
prompt=user_input,
execution=AgentExecutionConfig(max_retries_per_step=3, total_max_retries=10, max_iterations=20),
)
print(f"Received response: {response}")
# Create and store assistant's response
assistant_message = AssistantMessage(response.result.text)
# Await adding assistant message to memory
await memory.add(assistant_message)
print("Added assistant message to memory")
# Print results
print(f"\nMessages in memory: {len(agent.memory.messages)}")
if len(agent.memory.messages) >= 1:
user_msg = agent.memory.messages[0]
print(f"User: {user_msg.text}")
if len(agent.memory.messages) >= 2:
agent_msg = agent.memory.messages[1]
print(f"Agent: {agent_msg.text}")
else:
print("No agent message found in memory")
if __name__ == "__main__":
try:
asyncio.run(main())
except FrameworkError as e:
traceback.print_exc()
sys.exit(e.explain())
5. Event Observation
Monitor the agent’s execution by observing events it emits. This allows you to track its reasoning process, handle errors, or implement custom logging.
def update_callback(data: Any, event: EventMeta) -> None:
print(f"Agent({data.update.key}) 🤖 : ", data.update.parsed_value)
def on_update(emitter: Emitter) -> None:
emitter.on("update", update_callback)
output: BeeRunOutput = await agent.run("What's the current weather in Las Vegas?").observe(on_update)
Agent 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
