> ## Documentation Index
> Fetch the complete documentation index at: https://framework.beeai.dev/llms.txt
> Use this file to discover all available pages before exploring further.
# Agents
## 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.
Dive deeper into the concepts behind AI agents in this [research article](https://research.ibm.com/blog/what-are-ai-agents-llm) from IBM.
Supported in Python and TypeScript.
## Customizing Agent Behavior
You can customize your agent's behavior in several key ways:
### 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](./backend).
```py Python [expandable] theme={null}
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
)
```
```ts TypeScript [expandable] theme={null}
import { RequirementAgent } from "beeai-framework/agents/requirement/agent";
import { ChatModel } from "beeai-framework/backend/chat";
# Using Ollama (local models)
llm = await ChatModel.from_name("ollama:granite3.3")
# Using OpenAI
llm = await ChatModel.from_name("openai:gpt-5-mini")
# Using Anthropic
llm = await ChatModel.from_name("anthropic:claude-sonnet-4")
agent = new RequirementAgent({
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
The system prompt defines your agent's behavior, personality, and capabilities. You can configure this through several parameters when initializing an agent:
```py Python [expandable] theme={null}
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"
)
```
```ts TypeScript [expandable] theme={null}
const agent = new RequirementAgent({
llm: await ChatModel.from_name("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](./templates).
### 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
```py Python [expandable] theme={null}
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)
```
```ts TypeScript [expandable] theme={null}
const schema = z.object({
firstName: z.string().min(1),
lastName: z.string().min(1)
});
const response = await agent.run({
prompt: "Generate profile of a citizen.",
expectedOutput: schema,
});
const person = response.state.result
console.info(person.firstName, person.lastName);
```
**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
The are defaults set for `max_iterations`, `total_max_retries`, and `max_retries_per_step`, but you can override them by setting your own preferences.
### 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](./tools).
```py Python [expandable] theme={null}
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
]
)
```
```ts TypeScript [expandable] theme={null}
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { ChatModel } from "beeai-framework/backend/chat";
import { RequirementAgent } from "beeai-framework/agents/requirement/agent";
const llm = await ChatModel.from_name("ollama:granite4");
const agent = new RequirementAgent({
llm,
tools: [
new OpenMeteoTool(), // weather tool
],
});
```
### 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](./memory).
```py Python [expandable] theme={null}
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)
)
```
```ts TypeScript [expandable] theme={null}
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
import { ChatModel } from "beeai-framework/backend/chat";
import { RequirementAgent } from "beeai-framework/agents/requirement/agent";
const agent = new RequirementAgent({
llm: await ChatModel.from_name("ollama:granite4:micro"),
memory: new TokenMemory({ maxTokens: 20 * 1024 })
});
```
### Additional Agent Options
* [Observability & Debugging](./observability): Monitor agent behavior with detailed event tracking and logging systems
* [MCP (Model Context Protocol)](../integrations/mcp): Connect to external services and data sources
* [A2A (Agent-to-Agent)](../integrations/a2a): Enable multi-agent communication and coordination
* [Caching](./cache): Improve performance by caching LLM responses and tool outputs
* [Event System](./events): Build reactive applications using the comprehensive emitter framework
* [RAG Integration](./rag): Connect your agents to knowledge bases and document stores
* [Serialization](./serialization): Save and restore agent state for persistence and deployment
* [Error Handling](./errors): Implement robust error recovery and debugging strategies
## Agent Types
BeeAI Framework provides several agent implementations:
Upcoming change:
The Requirement agent will become the primary supported agent. The ReAct and tool-calling agents will not be actively supported.
### Requirement Agent
This is the recommended agent. Currently only supported in Python.
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](/modules/agents/requirement-agent) or in the [blog post](https://beeai.dev/blog/reliable-ai-agents).
```py Python [expandable] theme={null}
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
Currently only supported in Python.
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
```py Python [expandable] theme={null}
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
The ReAct Agent is available in both Python and TypeScript, but no longer actively supported.
The ReActAgent implements the ReAct ([Reasoning and Acting](https://arxiv.org/abs/2210.03629)) 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:**
```log theme={null}
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:**
```log theme={null}
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.
```py Python [expandable] theme={null}
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"
# "granite3.3"
# "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())
```
```ts TypeScript [expandable] theme={null}
import "dotenv/config.js";
import { ReActAgent } from "beeai-framework/agents/react/agent";
import { createConsoleReader } from "../helpers/io.js";
import { FrameworkError } from "beeai-framework/errors";
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
import { Logger } from "beeai-framework/logger/logger";
import { PythonTool } from "beeai-framework/tools/python/python";
import { LocalPythonStorage } from "beeai-framework/tools/python/storage";
import { DuckDuckGoSearchTool } from "beeai-framework/tools/search/duckDuckGoSearch";
import { WikipediaTool } from "beeai-framework/tools/search/wikipedia";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { dirname } from "node:path";
import { fileURLToPath } from "node:url";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";
Logger.root.level = "silent"; // disable internal logs
const logger = new Logger({ name: "app", level: "trace" });
// Other models to try:
// "llama3.1:70b"
// "granite3.3"
// "deepseek-r1:32b"
// ensure the model is pulled before running
const llm = new OllamaChatModel("granite4:micro");
const codeInterpreterUrl = process.env.CODE_INTERPRETER_URL;
const __dirname = dirname(fileURLToPath(import.meta.url));
const codeInterpreterTmpdir =
process.env.CODE_INTERPRETER_TMPDIR ?? "./examples/tmp/code_interpreter";
const localTmpdir = process.env.LOCAL_TMPDIR ?? "./examples/tmp/local";
const agent = new ReActAgent({
llm,
memory: new TokenMemory(),
tools: [
new DuckDuckGoSearchTool(),
// new WebCrawlerTool(), // HTML web page crawler
new WikipediaTool(),
new OpenMeteoTool(), // weather tool
// new ArXivTool(), // research papers
// new DynamicTool() // custom python tool
...(codeInterpreterUrl
? [
new PythonTool({
codeInterpreter: { url: codeInterpreterUrl },
storage: new LocalPythonStorage({
interpreterWorkingDir: `${__dirname}/../../${codeInterpreterTmpdir}`,
localWorkingDir: `${__dirname}/../../${localTmpdir}`,
}),
}),
]
: []),
],
});
const reader = createConsoleReader();
if (codeInterpreterUrl) {
reader.write(
"π οΈ System",
`The code interpreter tool is enabled. Please ensure that it is running on ${codeInterpreterUrl}`,
);
}
try {
for await (const { prompt } of reader) {
const response = await agent
.run(
{ prompt },
{
execution: {
maxRetriesPerStep: 3,
totalMaxRetries: 10,
maxIterations: 20,
},
},
)
.observe((emitter) => {
// emitter.on("start", () => {
// reader.write(`Agent π€ : `, "starting new iteration");
// });
emitter.on("error", ({ error }) => {
reader.write(`Agent π€ : `, FrameworkError.ensure(error).dump());
});
emitter.on("retry", () => {
reader.write(`Agent π€ : `, "retrying the action...");
});
emitter.on("update", async ({ data, update, meta }) => {
// log 'data' to see the whole state
// to log only valid runs (no errors), check if meta.success === true
reader.write(`Agent (${update.key}) π€ : `, update.value);
});
emitter.on("partialUpdate", ({ data, update, meta }) => {
// ideal for streaming (line by line)
// log 'data' to see the whole state
// to log only valid runs (no errors), check if meta.success === true
// reader.write(`Agent (partial ${update.key}) π€ : `, update.value);
});
// To observe all events (uncomment following block)
// emitter.match("*.*", async (data: unknown, event) => {
// logger.trace(event, `Received event "${event.path}"`);
// });
// To get raw LLM input (uncomment following block)
// emitter.match(
// (event) => event.creator === llm && event.name === "start",
// async (data: InferCallbackValue, event) => {
// logger.trace(
// event,
// [
// `Received LLM event "${event.path}"`,
// JSON.stringify(data.input), // array of messages
// ].join("\n"),
// );
// },
// );
});
reader.write(`Agent π€ : `, response.result.text);
}
} catch (error) {
logger.error(FrameworkError.ensure(error).dump());
} finally {
reader.close();
}
```
### Tool Calling Agent
The Tool Calling Agent is deprecated. Use [RequirementAgent](/modules/agents/requirement-agent) instead.
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.
```py Python [expandable] theme={null}
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())
```
```ts TypeScript [expandable] theme={null}
import "dotenv/config.js";
import { createConsoleReader } from "../../helpers/io.js";
import { FrameworkError } from "beeai-framework/errors";
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
import { Logger } from "beeai-framework/logger/logger";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";
import { ToolCallingAgent } from "beeai-framework/agents/toolCalling/agent";
Logger.root.level = "silent"; // disable internal logs
const logger = new Logger({ name: "app", level: "trace" });
// Other models to try:
// "llama3.1:70b"
// "granite3.3"
// "deepseek-r1:32b"
// ensure the model is pulled before running
const llm = new OllamaChatModel("granite4:micro");
const agent = new ToolCallingAgent({
llm,
memory: new TokenMemory(),
templates: {
system: (template) =>
template.fork((config) => {
config.defaults.instructions =
"You are a helpful assistant that uses tools to answer questions.";
}),
},
tools: [
new OpenMeteoTool(), // weather tool
],
});
const reader = createConsoleReader();
try {
for await (const { prompt } of reader) {
let messagesCount = agent.memory.messages.length + 1;
const response = await agent.run({ prompt }).observe((emitter) => {
emitter.on("success", async ({ state }) => {
const newMessages = state.memory.messages.slice(messagesCount);
messagesCount += newMessages.length;
reader.write(
`Agent (${newMessages.length} new messages) π€ :\n`,
newMessages.map((msg) => `-> ${JSON.stringify(msg.toPlain())}`).join("\n"),
);
});
// To observe all events (uncomment following block)
// emitter.match("*.*", async (data: unknown, event) => {
// logger.trace(event, `Received event "${event.path}"`);
// }, {
// matchNested: true
// });
// To get raw LLM input (uncomment following block)
// emitter.match(
// (event) => event.creator === llm && event.name === "start",
// async (data: InferCallbackValue, event) => {
// logger.trace(
// event,
// [
// `Received LLM event "${event.path}"`,
// JSON.stringify(data.input), // array of messages
// ].join("\n"),
// );
// },
// );
});
reader.write(`Agent π€ : `, response.result.text);
}
} catch (error) {
logger.error(FrameworkError.ensure(error).dump());
} finally {
reader.close();
}
```
### Custom Agent
For advanced use cases, you can create your own agent implementation by extending the `BaseAgent` class.
```py Python [expandable] theme={null}
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())
```
```ts TypeScript [expandable] theme={null}
import { BaseAgent, BaseAgentRunOptions } from "beeai-framework/agents/base";
import {
AssistantMessage,
Message,
SystemMessage,
UserMessage,
} from "beeai-framework/backend/message";
import { Emitter } from "beeai-framework/emitter/emitter";
import { GetRunContext } from "beeai-framework/context";
import { z } from "zod";
import { AgentMeta } from "beeai-framework/agents/types";
import { BaseMemory } from "beeai-framework/memory/base";
import { UnconstrainedMemory } from "beeai-framework/memory/unconstrainedMemory";
import { ChatModel } from "beeai-framework/backend/chat";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";
interface RunInput {
message: Message;
}
interface RunOutput {
message: Message;
state: {
thought: string;
final_answer: string;
};
}
interface RunOptions extends BaseAgentRunOptions {
maxRetries?: number;
}
interface AgentInput {
llm: ChatModel;
memory: BaseMemory;
}
export class CustomAgent extends BaseAgent {
public readonly memory: BaseMemory;
protected readonly model: ChatModel;
public emitter = Emitter.root.child({
namespace: ["agent", "custom"],
creator: this,
});
constructor(input: AgentInput) {
super();
this.model = input.llm;
this.memory = input.memory;
}
protected async _run(
input: RunInput,
options: RunOptions,
run: GetRunContext,
): Promise {
const response = await this.model.createStructure({
schema: z.object({
thought: z
.string()
.describe("Describe your thought process before coming with a final answer"),
final_answer: z
.string()
.describe("Here you should provide concise answer to the original question."),
}),
messages: [
new SystemMessage("You are a helpful assistant. Always use JSON format for you responses."),
...this.memory.messages,
input.message,
],
maxRetries: options?.maxRetries,
abortSignal: run.signal,
});
const result = new AssistantMessage(response.object.final_answer);
await this.memory.add(result);
return {
message: result,
state: response.object,
};
}
public get meta(): AgentMeta {
return {
name: "CustomAgent",
description: "Custom Agent is a simple LLM agent.",
tools: [],
};
}
createSnapshot() {
return {
...super.createSnapshot(),
emitter: this.emitter,
memory: this.memory,
};
}
loadSnapshot(snapshot: ReturnType) {
Object.assign(this, snapshot);
}
}
const agent = new CustomAgent({
llm: new OllamaChatModel("granite3.3"),
memory: new UnconstrainedMemory(),
});
const response = await agent.run({
message: new UserMessage("Why is the sky blue?"),
});
console.info(response.state);
```
## Multi-Agent Hand-offs
Create a team of specialized agents that can collaborate:
```py Python [expandable] theme={null}
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())
```
```ts TypeScript [expandable] theme={null}
COMING SOON
```
## Workflows
Upcoming change:
Workflows are under construction to support more dynamic multi-agent patterns. If you'd like to participate in shaping the vision, contribute to the discussion in this [V2 Workflow Proposal](https://github.com/i-am-bee/beeai-framework/discussions/1005).
For complex applications, you can create multi-agent workflows where specialized agents collaborate.
```py Python [expandable] theme={null}
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())
```
```ts TypeScript [expandable] theme={null}
import "dotenv/config";
import { createConsoleReader } from "examples/helpers/io.js";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { WikipediaTool } from "beeai-framework/tools/search/wikipedia";
import { AgentWorkflow } from "beeai-framework/workflows/agent";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";
const workflow = new AgentWorkflow("Smart assistant");
const llm = new OllamaChatModel("llama3.1");
workflow.addAgent({
name: "Researcher",
role: "A diligent researcher",
instructions: "You look up and provide information about a specific topic.",
tools: [new WikipediaTool()],
llm,
});
workflow.addAgent({
name: "WeatherForecaster",
role: "A weather reporter",
instructions: "You provide detailed weather reports.",
tools: [new OpenMeteoTool()],
llm,
});
workflow.addAgent({
name: "DataSynthesizer",
role: "A meticulous and creative data synthesizer",
instructions: "You can combine disparate information into a final coherent summary.",
llm,
});
const reader = createConsoleReader();
reader.write("Assistant π€ : ", "What location do you want to learn about?");
for await (const { prompt } of reader) {
const { result } = await workflow
.run([
{ prompt: "Provide a short history of the location.", context: prompt },
{
prompt: "Provide a comprehensive weather summary for the location today.",
expectedOutput:
"Essential weather details such as chance of rain, temperature and wind. Only report information that is available.",
},
{
prompt: "Summarize the historical and weather data for the location.",
expectedOutput:
"A paragraph that describes the history of the location, followed by the current weather conditions.",
},
])
.observe((emitter) => {
emitter.on("success", (data) => {
reader.write(
`Step '${data.step}' has been completed with the following outcome:\n`,
data.state?.finalAnswer ?? "-",
);
});
});
reader.write(`Assistant π€`, result.finalAnswer);
reader.write("Assistant π€ : ", "What location do you want to learn about?");
}
```
## Examples
Explore reference agent implementations in Python
Explore reference agent implementations in TypeScript