0
# Agent Framework
1
2
Build autonomous agents that can use tools, maintain conversation state, and execute complex multi-step workflows. Haystack's agent framework provides components for creating intelligent agents with function calling capabilities and persistent state management.
3
4
## Capabilities
5
6
### Agent Orchestration
7
8
Create intelligent agents that can reason, make decisions, and execute actions using available tools.
9
10
```python { .api }
11
class Agent:
12
def __init__(
13
self,
14
generator: Any,
15
tools: Optional[List[Tool]] = None,
16
toolset: Optional[Toolset] = None,
17
system_prompt: Optional[str] = None,
18
max_iterations: int = 10
19
) -> None:
20
"""
21
Initialize an autonomous agent.
22
23
Args:
24
generator: Chat generator component (e.g., OpenAIChatGenerator)
25
tools: List of individual tools available to the agent
26
toolset: Organized collection of tools
27
system_prompt: System instructions for the agent
28
max_iterations: Maximum number of reasoning iterations
29
"""
30
31
def run(
32
self,
33
messages: List[ChatMessage]
34
) -> Dict[str, List[ChatMessage]]:
35
"""
36
Run agent conversation with tool execution capabilities.
37
38
Args:
39
messages: List of chat messages (conversation history)
40
41
Returns:
42
Dictionary with 'messages' key containing updated conversation
43
"""
44
```
45
46
### Tool Invocation
47
48
Execute tools and functions called by language models with proper error handling and result formatting.
49
50
```python { .api }
51
class ToolInvoker:
52
def __init__(
53
self,
54
tools: Optional[List[Tool]] = None,
55
toolset: Optional[Toolset] = None,
56
raise_on_failure: bool = True
57
) -> None:
58
"""
59
Initialize tool invoker for executing function calls.
60
61
Args:
62
tools: List of tools available for invocation
63
toolset: Organized toolset for invocation
64
raise_on_failure: Whether to raise exceptions on tool failures
65
"""
66
67
def run(
68
self,
69
tool_calls: List[ToolCall]
70
) -> Dict[str, List[ToolCallResult]]:
71
"""
72
Execute tool calls and return results.
73
74
Args:
75
tool_calls: List of tool calls to execute
76
77
Returns:
78
Dictionary with 'tool_results' key containing execution results
79
"""
80
```
81
82
### Tool Management
83
84
Define, organize, and manage tools that agents can use.
85
86
```python { .api }
87
class Tool:
88
def __init__(
89
self,
90
name: str,
91
description: str,
92
function: Callable,
93
parameters: Dict[str, Any]
94
) -> None:
95
"""
96
Initialize a tool.
97
98
Args:
99
name: Unique tool name
100
description: Description of what the tool does
101
function: Python function to execute
102
parameters: JSON schema for function parameters
103
"""
104
105
@classmethod
106
def from_function(
107
cls,
108
function: Callable,
109
name: Optional[str] = None,
110
description: Optional[str] = None
111
) -> "Tool":
112
"""
113
Create tool from a Python function.
114
115
Args:
116
function: Python function to wrap as a tool
117
name: Optional custom name (defaults to function name)
118
description: Optional custom description (defaults to docstring)
119
120
Returns:
121
Tool instance wrapping the function
122
"""
123
124
def invoke(self, **kwargs) -> Any:
125
"""
126
Invoke the tool with given arguments.
127
128
Args:
129
**kwargs: Arguments to pass to the tool function
130
131
Returns:
132
Result of tool execution
133
"""
134
135
@tool
136
def my_custom_tool(param1: str, param2: int = 10) -> str:
137
"""
138
Decorator to create a tool from a function.
139
140
Args:
141
param1: First parameter description
142
param2: Second parameter with default value
143
144
Returns:
145
Tool execution result
146
"""
147
148
class Toolset:
149
def __init__(
150
self,
151
tools: List[Tool]
152
) -> None:
153
"""
154
Initialize a collection of tools.
155
156
Args:
157
tools: List of tools to include in the toolset
158
"""
159
160
def add_tool(self, tool: Tool) -> None:
161
"""Add a tool to the toolset."""
162
163
def get_tool(self, name: str) -> Optional[Tool]:
164
"""Get a tool by name."""
165
166
def list_tools(self) -> List[str]:
167
"""List all tool names."""
168
```
169
170
### State Management
171
172
Manage agent state and conversation context across multiple interactions.
173
174
```python { .api }
175
class State:
176
def __init__(
177
self,
178
data: Optional[Dict[str, Any]] = None
179
) -> None:
180
"""
181
Initialize agent state.
182
183
Args:
184
data: Initial state data dictionary
185
"""
186
187
def get(self, key: str, default: Any = None) -> Any:
188
"""
189
Get value from state.
190
191
Args:
192
key: State key to retrieve
193
default: Default value if key not found
194
195
Returns:
196
State value or default
197
"""
198
199
def set(self, key: str, value: Any) -> None:
200
"""
201
Set state value.
202
203
Args:
204
key: State key to set
205
value: Value to store
206
"""
207
208
def update(self, data: Dict[str, Any]) -> None:
209
"""
210
Update state with multiple key-value pairs.
211
212
Args:
213
data: Dictionary of state updates
214
"""
215
216
def clear(self) -> None:
217
"""Clear all state data."""
218
219
def to_dict(self) -> Dict[str, Any]:
220
"""Convert state to dictionary."""
221
```
222
223
## Usage Examples
224
225
### Basic Agent with Tools
226
227
```python
228
from haystack.components.agents import Agent
229
from haystack.components.generators.chat import OpenAIChatGenerator
230
from haystack.tools import Tool
231
from haystack.dataclasses import ChatMessage, ChatRole
232
from haystack.utils import Secret
233
import requests
234
from datetime import datetime
235
236
# Define tools for the agent
237
@tool
238
def get_weather(city: str) -> str:
239
"""
240
Get current weather for a city.
241
242
Args:
243
city: Name of the city
244
245
Returns:
246
Weather description
247
"""
248
# Simulate weather API call
249
return f"Weather in {city}: Sunny, 22°C"
250
251
@tool
252
def get_current_time() -> str:
253
"""
254
Get current date and time.
255
256
Returns:
257
Current timestamp
258
"""
259
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
260
261
# Create tools list
262
tools = [get_weather, get_current_time]
263
264
# Initialize chat generator
265
chat_generator = OpenAIChatGenerator(
266
api_key=Secret.from_env_var("OPENAI_API_KEY"),
267
model="gpt-3.5-turbo",
268
tools=tools
269
)
270
271
# Create agent
272
agent = Agent(
273
generator=chat_generator,
274
tools=tools,
275
system_prompt="You are a helpful assistant that can check weather and time."
276
)
277
278
# Run agent conversation
279
messages = [
280
ChatMessage(
281
content="What's the weather like in Paris and what time is it now?",
282
role=ChatRole.USER
283
)
284
]
285
286
result = agent.run(messages=messages)
287
for message in result["messages"]:
288
print(f"{message.role.value}: {message.content}")
289
```
290
291
### Complex Multi-Tool Agent
292
293
```python
294
from haystack.tools import Toolset
295
import json
296
import sqlite3
297
298
# Define a more complex toolset
299
@tool
300
def search_database(query: str, table: str) -> str:
301
"""
302
Search database for information.
303
304
Args:
305
query: SQL-like search query
306
table: Database table to search
307
308
Returns:
309
JSON string with search results
310
"""
311
# Simulate database search
312
results = [
313
{"id": 1, "name": "John Doe", "email": "john@example.com"},
314
{"id": 2, "name": "Jane Smith", "email": "jane@example.com"}
315
]
316
return json.dumps(results)
317
318
@tool
319
def send_email(to: str, subject: str, body: str) -> str:
320
"""
321
Send an email.
322
323
Args:
324
to: Recipient email address
325
subject: Email subject
326
body: Email body content
327
328
Returns:
329
Confirmation message
330
"""
331
# Simulate email sending
332
return f"Email sent to {to} with subject '{subject}'"
333
334
@tool
335
def calculate_math(expression: str) -> str:
336
"""
337
Calculate mathematical expressions.
338
339
Args:
340
expression: Mathematical expression to evaluate
341
342
Returns:
343
Calculation result
344
"""
345
try:
346
result = eval(expression) # In production, use safe evaluation
347
return str(result)
348
except Exception as e:
349
return f"Error: {str(e)}"
350
351
# Create toolset
352
business_toolset = Toolset([
353
search_database,
354
send_email,
355
calculate_math
356
])
357
358
# Create business agent
359
business_agent = Agent(
360
generator=OpenAIChatGenerator(
361
api_key=Secret.from_env_var("OPENAI_API_KEY"),
362
model="gpt-4",
363
tools=business_toolset.tools
364
),
365
toolset=business_toolset,
366
system_prompt="""
367
You are a business assistant that can:
368
1. Search the company database
369
2. Send emails to contacts
370
3. Perform calculations
371
372
Always be professional and helpful.
373
"""
374
)
375
376
# Complex multi-step task
377
messages = [
378
ChatMessage(
379
content="Find John Doe in the database, calculate 15% of 1000, and send him an email about the discount.",
380
role=ChatRole.USER
381
)
382
]
383
384
result = business_agent.run(messages=messages)
385
```
386
387
### Agent with State Management
388
389
```python
390
from haystack.components.agents import State
391
392
# Create stateful agent
393
class StatefulAgent:
394
def __init__(self, agent: Agent):
395
self.agent = agent
396
self.state = State()
397
self.conversation_history = []
398
399
def chat(self, user_input: str) -> str:
400
# Add user input to conversation
401
user_message = ChatMessage(content=user_input, role=ChatRole.USER)
402
self.conversation_history.append(user_message)
403
404
# Update state with conversation count
405
conversation_count = self.state.get("conversation_count", 0) + 1
406
self.state.set("conversation_count", conversation_count)
407
408
# Add context from state
409
context_message = ChatMessage(
410
content=f"This is conversation #{conversation_count}. "
411
f"Previous context: {self.state.to_dict()}",
412
role=ChatRole.SYSTEM
413
)
414
415
# Run agent with full context
416
messages_with_context = [context_message] + self.conversation_history
417
result = self.agent.run(messages=messages_with_context)
418
419
# Extract and store assistant response
420
assistant_response = result["messages"][-1]
421
self.conversation_history.append(assistant_response)
422
423
# Update state with last response summary
424
self.state.set("last_topic", user_input[:50])
425
426
return assistant_response.content
427
428
# Usage
429
stateful_agent = StatefulAgent(agent)
430
431
response1 = stateful_agent.chat("My name is Alice")
432
response2 = stateful_agent.chat("What's my name?") # Should remember Alice
433
response3 = stateful_agent.chat("What did we talk about before?") # Should reference history
434
```
435
436
### Tool Invoker Pipeline
437
438
```python
439
from haystack.components.tools import ToolInvoker
440
from haystack import Pipeline
441
from haystack.dataclasses import ToolCall
442
443
# Create tool invocation pipeline
444
tool_pipeline = Pipeline()
445
446
# Add tool invoker
447
tool_invoker = ToolInvoker(
448
tools=[get_weather, get_current_time],
449
raise_on_failure=False
450
)
451
452
tool_pipeline.add_component("tool_invoker", tool_invoker)
453
454
# Create tool calls
455
tool_calls = [
456
ToolCall(
457
tool_name="get_weather",
458
arguments={"city": "London"},
459
id="call_1"
460
),
461
ToolCall(
462
tool_name="get_current_time",
463
arguments={},
464
id="call_2"
465
)
466
]
467
468
# Execute tools
469
result = tool_pipeline.run({
470
"tool_invoker": {"tool_calls": tool_calls}
471
})
472
473
# Process results
474
for tool_result in result["tool_invoker"]["tool_results"]:
475
print(f"Tool {tool_result.origin.tool_name}: {tool_result.result}")
476
if tool_result.error:
477
print(f"Error occurred: {tool_result.error}")
478
```
479
480
### Advanced Agent with Error Handling
481
482
```python
483
@tool
484
def risky_operation(value: int) -> str:
485
"""
486
Perform an operation that might fail.
487
488
Args:
489
value: Input value
490
491
Returns:
492
Operation result or error message
493
"""
494
if value < 0:
495
raise ValueError("Value must be positive")
496
return f"Success: {value * 2}"
497
498
# Create agent with error handling
499
class RobustAgent:
500
def __init__(self, agent: Agent):
501
self.agent = agent
502
self.error_count = 0
503
504
def safe_run(self, messages: List[ChatMessage]) -> Dict[str, Any]:
505
try:
506
result = self.agent.run(messages=messages)
507
return {"success": True, "result": result}
508
except Exception as e:
509
self.error_count += 1
510
return {
511
"success": False,
512
"error": str(e),
513
"error_count": self.error_count
514
}
515
516
# Usage with error handling
517
robust_agent = RobustAgent(
518
Agent(
519
generator=chat_generator,
520
tools=[risky_operation],
521
system_prompt="Handle errors gracefully and inform the user."
522
)
523
)
524
525
messages = [
526
ChatMessage(
527
content="Try risky_operation with value -5",
528
role=ChatRole.USER
529
)
530
]
531
532
result = robust_agent.safe_run(messages)
533
if result["success"]:
534
print("Agent succeeded:", result["result"])
535
else:
536
print("Agent failed:", result["error"])
537
```
538
539
### Agent Workflow with Multiple Steps
540
541
```python
542
# Create workflow agent
543
class WorkflowAgent:
544
def __init__(self):
545
self.steps = []
546
self.current_step = 0
547
self.context = {}
548
549
def add_step(self, step_name: str, agent: Agent):
550
self.steps.append({"name": step_name, "agent": agent})
551
552
def execute_workflow(self, initial_input: str) -> Dict[str, Any]:
553
results = {}
554
current_input = initial_input
555
556
for i, step in enumerate(self.steps):
557
print(f"Executing step {i+1}: {step['name']}")
558
559
messages = [
560
ChatMessage(
561
content=f"Context: {self.context}\nTask: {current_input}",
562
role=ChatRole.USER
563
)
564
]
565
566
step_result = step["agent"].run(messages=messages)
567
results[step["name"]] = step_result
568
569
# Extract output for next step
570
assistant_message = step_result["messages"][-1]
571
current_input = assistant_message.content
572
573
# Update context
574
self.context[step["name"]] = {
575
"input": messages[0].content,
576
"output": current_input
577
}
578
579
return results
580
581
# Create workflow
582
workflow = WorkflowAgent()
583
584
# Add workflow steps
585
workflow.add_step("research", Agent(
586
generator=chat_generator,
587
tools=[search_database],
588
system_prompt="Research the given topic and provide key information."
589
))
590
591
workflow.add_step("analysis", Agent(
592
generator=chat_generator,
593
tools=[calculate_math],
594
system_prompt="Analyze the research data and provide insights."
595
))
596
597
workflow.add_step("communication", Agent(
598
generator=chat_generator,
599
tools=[send_email],
600
system_prompt="Create a professional communication based on the analysis."
601
))
602
603
# Execute workflow
604
workflow_results = workflow.execute_workflow("Analyze customer satisfaction data")
605
```
606
607
### Multi-Agent Conversation
608
609
```python
610
# Create multiple specialized agents
611
class MultiAgentSystem:
612
def __init__(self):
613
self.agents = {}
614
615
def add_agent(self, name: str, agent: Agent):
616
self.agents[name] = agent
617
618
def facilitate_conversation(self, topic: str, max_turns: int = 5) -> List[Dict[str, str]]:
619
conversation = []
620
current_speaker = list(self.agents.keys())[0]
621
current_message = f"Let's discuss: {topic}"
622
623
for turn in range(max_turns):
624
# Get current agent
625
agent = self.agents[current_speaker]
626
627
# Add conversation history to context
628
context = "\n".join([
629
f"{entry['speaker']}: {entry['message']}"
630
for entry in conversation[-3:] # Last 3 exchanges
631
])
632
633
messages = [
634
ChatMessage(
635
content=f"Context: {context}\n\nRespond to: {current_message}",
636
role=ChatRole.USER
637
)
638
]
639
640
result = agent.run(messages=messages)
641
response = result["messages"][-1].content
642
643
conversation.append({
644
"speaker": current_speaker,
645
"message": response,
646
"turn": turn + 1
647
})
648
649
# Switch to next agent
650
agent_names = list(self.agents.keys())
651
current_index = agent_names.index(current_speaker)
652
current_speaker = agent_names[(current_index + 1) % len(agent_names)]
653
current_message = response
654
655
return conversation
656
657
# Create multi-agent system
658
mas = MultiAgentSystem()
659
660
mas.add_agent("researcher", Agent(
661
generator=chat_generator,
662
system_prompt="You are a researcher. Focus on facts and data."
663
))
664
665
mas.add_agent("critic", Agent(
666
generator=chat_generator,
667
system_prompt="You are a critic. Challenge assumptions and ask hard questions."
668
))
669
670
mas.add_agent("synthesizer", Agent(
671
generator=chat_generator,
672
system_prompt="You synthesize different viewpoints into coherent conclusions."
673
))
674
675
# Run multi-agent conversation
676
conversation = mas.facilitate_conversation("The impact of AI on education", max_turns=6)
677
678
for entry in conversation:
679
print(f"{entry['speaker']} (Turn {entry['turn']}): {entry['message']}")
680
```
681
682
## Types
683
684
```python { .api }
685
from typing import Callable, Dict, Any, List, Optional, Union
686
from haystack.dataclasses import ChatMessage, ToolCall, ToolCallResult
687
688
class AgentStep:
689
"""Represents a single step in agent reasoning."""
690
thought: str
691
action: Optional[ToolCall]
692
observation: Optional[str]
693
694
class AgentTrace:
695
"""Complete trace of agent execution."""
696
steps: List[AgentStep]
697
final_answer: str
698
total_time: float
699
700
class ToolDefinition:
701
"""Tool definition for function calling."""
702
name: str
703
description: str
704
parameters: Dict[str, Any]
705
function: Callable
706
```