Xây Dựng Agent Đầu Tiên: Instructions, Tools và Function Calling
Hướng dẫn chi tiết cách viết instructions hiệu quả và tạo function tools cho OpenAI Agents. Từ basic function calling đến advanced tool patterns.
Trong bài trước, chúng ta đã làm quen với những khái niệm cơ bản của OpenAI Agents SDK. Nhưng để tạo ra những agents thực sự hữu ích, chúng ta cần trang bị cho chúng khả năng thực hiện hành động thông qua tools và hiểu rõ nhiệm vụ thông qua instructions hiệu quả.
Hôm nay, chúng ta sẽ khám phá cách biến agents từ những “chatbot đơn thuần” thành những “digital workers” có thể tương tác với APIs, xử lý dữ liệu, và thực hiện các tác vụ phức tạp trong thế giới thực.
Khái niệm cơ bản
Instructions
Instructions (System prompt) là tập hợp các hướng dẫn định nghĩa nhiệm vụ, vai trò, và giới hạn cho một Agent. Instructions giúp Agent hiểu rõ mục tiêu và cách thức phản hồi, từ đó đảm bảo kết quả nhất quán và chính xác hơn.
Tools
Tools là các chức năng hoặc phương thức được cung cấp để Agent tương tác với thế giới bên ngoài như lấy thông tin, xử lý dữ liệu, hay thực hiện các hành động cụ thể. Ví dụ: lấy thông tin thời tiết, gửi email, tính toán dữ liệu…
Tại Sao Instructions và Tools Quan Trọng?
Instructions - Bộ Não Của Agent
Instructions (hay system prompt) không chỉ là “hướng dẫn sử dụng” mà còn là DNA của agent - định nghĩa personality, capabilities, và behavior patterns. Một instruction tốt có thể:
- Định hướng hành vi: Agent sẽ phản ứng như thế nào trong các tình huống khác nhau
- Thiết lập context: Vai trò, expertise, và domain knowledge
- Đặt ra boundaries: Những gì agent nên và không nên làm
- Cải thiện accuracy: Giảm hallucination và tăng tính nhất quán
Tools - Đôi Tay Của Agent
Tools cho phép agents:
- Truy cập dữ liệu real-time: Weather, news, stock prices
- Tương tác với services: Send emails, create calendar events
- Xử lý files: Read/write documents, analyze spreadsheets
- Thực hiện calculations: Complex math, data analysis
- Control systems: Automate workflows, manage infrastructure
Viết Instructions Hiệu Quả
Cấu Trúc Instructions Tối Ưu
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def create_expert_instructions(domain: str, role: str, constraints: list = None) -> str:
"""Template tạo instructions chuyên nghiệp"""
constraints = constraints or []
constraints_text = "\n".join([f"- {c}" for c in constraints])
return f"""
# Role & Identity
You are a {role} specializing in {domain}.
# Core Responsibilities
- Provide accurate, actionable advice
- Ask clarifying questions when requests are ambiguous
- Explain your reasoning step-by-step
- Admit when you're uncertain about something
# Communication Style
- Be professional yet approachable
- Use clear, concise language
- Provide examples when helpful
- Structure responses with headers and bullet points
# Constraints & Guidelines
{constraints_text}
# Quality Standards
- Always double-check facts and calculations
- Cite sources when making specific claims
- Focus on practical, implementable solutions
- Consider potential risks and alternatives
"""
# Ví dụ sử dụng
financial_advisor_instructions = create_expert_instructions(
domain="personal finance and investment",
role="Senior Financial Advisor",
constraints=[
"Never provide specific stock recommendations",
"Always mention risks associated with investments",
"Suggest consulting with certified professionals for major decisions",
"Focus on educational content rather than direct advice"
]
)
Dynamic Instructions - Thông Minh Hơn Với Context
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from datetime import datetime
from agents import Agent, RunContextWrapper
def dynamic_market_instructions(
context: RunContextWrapper,
agent: Agent
) -> str:
"""Instructions thay đổi theo thời gian và context"""
current_time = datetime.now()
market_hours = 9 <= current_time.hour <= 16
is_weekend = current_time.weekday() >= 5
base_instructions = """
You are a Financial Market Assistant with real-time market awareness.
Your capabilities:
- Analyze market trends and patterns
- Explain financial concepts clearly
- Provide educational market insights
"""
# Thêm context về thời gian
time_context = f"""
# Current Context
- Current time: {current_time.strftime('%Y-%m-%d %H:%M %Z')}
- Market status: {'OPEN' if market_hours and not is_weekend else 'CLOSED'}
"""
if market_hours and not is_weekend:
time_context += """
- Focus on real-time market movements
- Mention that data might be delayed
- Emphasize volatility during trading hours
"""
else:
time_context += """
- Focus on market analysis and education
- Mention markets are currently closed
- Discuss preparation for next trading session
"""
return base_instructions + time_context
# Sử dụng dynamic instructions
market_agent = Agent(
name="Market Assistant",
instructions=dynamic_market_instructions,
# tools sẽ được thêm sau
)
Best Practices Cho Instructions
✅ DO - Những Điều Nên Làm:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
good_instructions = """
# Role Definition
You are a Code Review Assistant for Python projects.
# Specific Tasks
1. Analyze code for bugs, security issues, and performance problems
2. Suggest improvements following PEP 8 and best practices
3. Explain the reasoning behind each suggestion
4. Provide code examples for recommended changes
# Response Format
For each file reviewed:
## Summary
- Overall code quality: [Score/10]
- Main issues found: [count]
## Detailed Feedback
### Critical Issues (Security/Bugs)
- [List critical issues with line numbers]
### Improvements (Performance/Style)
- [List improvement suggestions]
### Positive Aspects
- [Highlight good practices found]
# Tone & Style
- Be constructive and encouraging
- Focus on education, not just criticism
- Provide rationale for suggestions
- Use technical terms appropriately
"""
❌ DON’T - Những Điều Tránh:
1
2
3
4
5
6
bad_instructions = """
Help with code. # Too vague
Be smart. # Meaningless
Don't be wrong. # Negative framing
Do everything. # Unrealistic scope
"""
Function Tools - Sức Mạnh Thật Sự
Cách Hoạt Động Của Function Tools
OpenAI Agents SDK tự động:
- Extract function signature từ Python function
- Generate JSON schema cho parameters
- Parse docstring để lấy descriptions
- Handle type validation với Pydantic
- Execute function khi LLM gọi tool
Basic Function Tools
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
import requests
from typing import Dict, Any
from agents import Agent, Runner, function_tool
@function_tool
def get_weather(city: str, country_code: str = "US") -> str:
"""Get current weather information for a specific city.
Args:
city: Name of the city (e.g., "New York", "London")
country_code: ISO country code (e.g., "US", "UK", "VN")
Returns:
Weather information as a formatted string
"""
# Demo implementation - trong thực tế sẽ call real weather API
try:
# Mock weather data
weather_data = {
"temperature": 22,
"condition": "Sunny",
"humidity": 65,
"wind_speed": 10
}
return f"""
Weather in {city}, {country_code}:
🌡️ Temperature: {weather_data['temperature']}°C
☀️ Condition: {weather_data['condition']}
💧 Humidity: {weather_data['humidity']}%
💨 Wind Speed: {weather_data['wind_speed']} km/h
""".strip()
except Exception as e:
return f"Sorry, I couldn't fetch weather data for {city}. Error: {str(e)}"
@function_tool
def calculate_compound_interest(
principal: float,
annual_rate: float,
years: int,
compound_frequency: int = 12
) -> Dict[str, Any]:
"""Calculate compound interest over time.
Args:
principal: Initial investment amount in dollars
annual_rate: Annual interest rate as decimal (e.g., 0.05 for 5%)
years: Number of years to calculate
compound_frequency: How many times per year interest compounds (default: 12 for monthly)
Returns:
Dictionary with calculation results
"""
if principal <= 0 or annual_rate < 0 or years <= 0:
return {"error": "Invalid input parameters"}
# A = P(1 + r/n)^(nt)
final_amount = principal * (1 + annual_rate/compound_frequency) ** (compound_frequency * years)
total_interest = final_amount - principal
return {
"principal": principal,
"annual_rate": annual_rate * 100, # Convert to percentage
"years": years,
"compound_frequency": compound_frequency,
"final_amount": round(final_amount, 2),
"total_interest": round(total_interest, 2),
"growth_percentage": round((total_interest / principal) * 100, 2)
}
# Tạo agent với tools
weather_finance_agent = Agent(
name="Weather & Finance Assistant",
instructions="""
You are a helpful assistant that can provide weather information and financial calculations.
Capabilities:
- Get current weather for any city
- Calculate compound interest for investments
When providing financial calculations:
- Always explain the assumptions used
- Mention that this is for educational purposes
- Suggest consulting with financial advisors for actual investment decisions
Format your responses clearly with appropriate emojis and structure.
""",
tools=[get_weather, calculate_compound_interest]
)
# Test the agent
async def test_tools():
# Weather query
result1 = await Runner.run(
weather_finance_agent,
"What's the weather like in Hanoi, Vietnam?"
)
print("Weather Query Result:")
print(result1.final_output)
print("\n" + "="*50 + "\n")
# Finance query
result2 = await Runner.run(
weather_finance_agent,
"If I invest $10,000 at 7% annual interest for 20 years, how much will I have?"
)
print("Finance Query Result:")
print(result2.final_output)
if __name__ == "__main__":
import asyncio
asyncio.run(test_tools())
Advanced Function Tools với Context
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from dataclasses import dataclass
from typing import List, Optional
from agents import RunContextWrapper, function_tool
@dataclass
class UserProfile:
user_id: str
name: str
email: str
preferences: Dict[str, Any]
subscription_level: str = "basic"
@function_tool
async def get_user_preferences(
ctx: RunContextWrapper[UserProfile],
category: str
) -> Dict[str, Any]:
"""Get user preferences for a specific category.
Args:
category: Category of preferences (e.g., 'notifications', 'display', 'privacy')
Returns:
User preferences for the specified category
"""
user = ctx.context
# Simulate database lookup
all_preferences = {
"notifications": {
"email": user.subscription_level != "basic",
"push": True,
"frequency": "daily" if user.subscription_level == "premium" else "weekly"
},
"display": {
"theme": "dark",
"language": "en",
"timezone": "Asia/Ho_Chi_Minh"
},
"privacy": {
"analytics": user.subscription_level == "premium",
"data_sharing": False
}
}
return {
"user_id": user.user_id,
"category": category,
"preferences": all_preferences.get(category, {}),
"subscription_level": user.subscription_level
}
@function_tool
async def update_user_preferences(
ctx: RunContextWrapper[UserProfile],
category: str,
updates: Dict[str, Any]
) -> str:
"""Update user preferences for a specific category.
Args:
category: Category to update
updates: Dictionary of preference updates
Returns:
Confirmation message
"""
user = ctx.context
# Validate subscription level for premium features
if category == "notifications" and updates.get("frequency") == "realtime":
if user.subscription_level != "premium":
return "Real-time notifications require a Premium subscription. Please upgrade to access this feature."
# Simulate database update
print(f"Updating {category} preferences for user {user.name}")
print(f"Updates: {updates}")
return f"Successfully updated {category} preferences for {user.name}. Changes will take effect immediately."
# Context-aware agent
personalized_agent = Agent[UserProfile](
name="Personal Settings Assistant",
instructions="""
You are a personal settings assistant that helps users manage their account preferences.
You have access to the user's profile and can:
- Retrieve current preferences by category
- Update preferences based on user requests
- Explain subscription-level restrictions
Always:
- Address the user by name
- Explain what changes were made
- Mention subscription limitations when relevant
- Ask for confirmation before making changes
""",
tools=[get_user_preferences, update_user_preferences]
)
# Usage example
async def demo_context_tools():
# Create user profile
user = UserProfile(
user_id="user_123",
name="Alice",
email="alice@example.com",
preferences={},
subscription_level="basic"
)
# Test the agent
result = await Runner.run(
personalized_agent,
"Can you show me my notification settings and help me enable real-time notifications?",
context=user
)
print(result.final_output)
if __name__ == "__main__":
import asyncio
asyncio.run(demo_context_tools())
Custom Function Tools - Kiểm Soát Hoàn Toàn
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
from pydantic import BaseModel
from agents import FunctionTool, RunContextWrapper
class DatabaseQuery(BaseModel):
table: str
where_clause: Optional[str] = None
limit: int = 10
async def execute_safe_query(ctx: RunContextWrapper, args: str) -> str:
"""Execute a database query with safety checks"""
try:
query_params = DatabaseQuery.model_validate_json(args)
# Safety checks
dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'TRUNCATE']
if query_params.where_clause:
if any(keyword in query_params.where_clause.upper() for keyword in dangerous_keywords):
return "Error: Dangerous SQL keywords detected. Only SELECT queries are allowed."
# Simulate database query
mock_results = [
{"id": 1, "name": "Product A", "price": 29.99},
{"id": 2, "name": "Product B", "price": 49.99},
]
return f"Query Results from {query_params.table}:\n" + \
"\n".join([str(row) for row in mock_results[:query_params.limit]])
except Exception as e:
return f"Query execution error: {str(e)}"
# Create custom tool
database_tool = FunctionTool(
name="query_database",
description="Execute safe SELECT queries on the database",
params_json_schema=DatabaseQuery.model_json_schema(),
on_invoke_tool=execute_safe_query,
)
# Agent with custom tool
database_agent = Agent(
name="Database Assistant",
instructions="""
You are a database assistant that can help users query data safely.
Security restrictions:
- Only SELECT queries are allowed
- No modification operations (INSERT, UPDATE, DELETE, DROP)
- Query results are limited to prevent overload
Always explain what data you're retrieving and why.
""",
tools=[database_tool]
)
Error Handling trong Function Tools
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from agents import function_tool
@function_tool(failure_error_function=None) # Re-raise errors for custom handling
def risky_api_call(endpoint: str) -> str:
"""Call external API with proper error handling.
Args:
endpoint: API endpoint to call
Returns:
API response data
"""
try:
# Simulate API call
if "invalid" in endpoint:
raise ValueError("Invalid endpoint format")
if "timeout" in endpoint:
raise TimeoutError("API request timed out")
return f"Success: Data from {endpoint}"
except ValueError as e:
return f"Validation Error: {str(e)}"
except TimeoutError as e:
return f"Timeout Error: {str(e)}. Please try again later."
except Exception as e:
return f"Unexpected Error: {str(e)}"
def custom_error_handler(error: Exception, tool_name: str, args: str) -> str:
"""Custom error handler for tools"""
return f"The {tool_name} tool encountered an issue. Please try rephrasing your request or contact support if the problem persists."
@function_tool(failure_error_function=custom_error_handler)
def another_tool(data: str) -> str:
"""Another tool with custom error handling"""
if not data:
raise ValueError("Data cannot be empty")
return f"Processed: {data}"
Real-World Example: Smart Task Manager
Hãy xây dựng một task manager thông minh với đầy đủ tính năng:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from enum import Enum
from pydantic import BaseModel
from agents import Agent, Runner, function_tool
class Priority(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
URGENT = "urgent"
class TaskStatus(str, Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
CANCELLED = "cancelled"
class Task(BaseModel):
id: str
title: str
description: Optional[str] = None
priority: Priority = Priority.MEDIUM
status: TaskStatus = TaskStatus.PENDING
due_date: Optional[str] = None
created_at: str
tags: List[str] = []
# In-memory storage (trong production sẽ dùng database)
tasks_storage: Dict[str, Task] = {}
@function_tool
def create_task(
title: str,
description: str = "",
priority: str = "medium",
due_date: str = "",
tags: str = ""
) -> str:
"""Create a new task.
Args:
title: Task title
description: Detailed description
priority: Task priority (low, medium, high, urgent)
due_date: Due date in YYYY-MM-DD format
tags: Comma-separated tags
Returns:
Confirmation message with task ID
"""
try:
task_id = f"task_{len(tasks_storage) + 1}"
tag_list = [tag.strip() for tag in tags.split(",") if tag.strip()]
task = Task(
id=task_id,
title=title,
description=description,
priority=Priority(priority.lower()),
due_date=due_date if due_date else None,
created_at=datetime.now().isoformat(),
tags=tag_list
)
tasks_storage[task_id] = task
return f"""✅ Task created successfully!
**Task ID:** {task_id}
**Title:** {title}
**Priority:** {priority.upper()}
**Due Date:** {due_date or 'Not set'}
**Tags:** {', '.join(tag_list) if tag_list else 'None'}
You can reference this task using ID: {task_id}"""
except Exception as e:
return f"❌ Error creating task: {str(e)}"
@function_tool
def list_tasks(
status: str = "all",
priority: str = "all",
limit: int = 10
) -> str:
"""List tasks with optional filtering.
Args:
status: Filter by status (all, pending, in_progress, completed, cancelled)
priority: Filter by priority (all, low, medium, high, urgent)
limit: Maximum number of tasks to return
Returns:
Formatted list of tasks
"""
try:
filtered_tasks = list(tasks_storage.values())
# Filter by status
if status != "all":
filtered_tasks = [t for t in filtered_tasks if t.status.value == status]
# Filter by priority
if priority != "all":
filtered_tasks = [t for t in filtered_tasks if t.priority.value == priority]
# Sort by priority and due date
priority_order = {"urgent": 4, "high": 3, "medium": 2, "low": 1}
filtered_tasks.sort(
key=lambda t: (
priority_order.get(t.priority.value, 0),
t.due_date or "9999-12-31"
),
reverse=True
)
if not filtered_tasks:
return "📝 No tasks found matching your criteria."
result = f"📋 **Task List** (showing {min(len(filtered_tasks), limit)} of {len(filtered_tasks)} tasks)\n\n"
for task in filtered_tasks[:limit]:
priority_emoji = {
"urgent": "🔥",
"high": "⚡",
"medium": "📝",
"low": "💤"
}
status_emoji = {
"pending": "⏳",
"in_progress": "🔄",
"completed": "✅",
"cancelled": "❌"
}
due_info = f" (Due: {task.due_date})" if task.due_date else ""
tags_info = f" #{', #'.join(task.tags)}" if task.tags else ""
result += f"{priority_emoji.get(task.priority.value, '📝')} **{task.title}** [{task.id}]\n"
result += f"{status_emoji.get(task.status.value, '⏳')} {task.status.value.replace('_', ' ').title()}{due_info}\n"
if task.description:
result += f"💭 {task.description}\n"
if tags_info:
result += f"🏷️ {tags_info}\n"
result += "\n"
return result.strip()
except Exception as e:
return f"❌ Error listing tasks: {str(e)}"
@function_tool
def update_task_status(task_id: str, new_status: str) -> str:
"""Update task status.
Args:
task_id: Task ID to update
new_status: New status (pending, in_progress, completed, cancelled)
Returns:
Confirmation message
"""
try:
if task_id not in tasks_storage:
return f"❌ Task {task_id} not found."
old_status = tasks_storage[task_id].status.value
tasks_storage[task_id].status = TaskStatus(new_status.lower())
status_emoji = {
"pending": "⏳",
"in_progress": "🔄",
"completed": "✅",
"cancelled": "❌"
}
return f"{status_emoji.get(new_status.lower(), '📝')} Task **{tasks_storage[task_id].title}** status updated from {old_status} to {new_status}!"
except Exception as e:
return f"❌ Error updating task: {str(e)}"
@function_tool
def get_task_analytics() -> str:
"""Get task analytics and insights.
Returns:
Task statistics and insights
"""
try:
if not tasks_storage:
return "📊 No tasks available for analysis."
total_tasks = len(tasks_storage)
status_counts = {}
priority_counts = {}
overdue_count = 0
today = datetime.now().date()
for task in tasks_storage.values():
# Count by status
status_counts[task.status.value] = status_counts.get(task.status.value, 0) + 1
# Count by priority
priority_counts[task.priority.value] = priority_counts.get(task.priority.value, 0) + 1
# Check overdue
if task.due_date and task.status != TaskStatus.COMPLETED:
due_date = datetime.strptime(task.due_date, "%Y-%m-%d").date()
if due_date < today:
overdue_count += 1
completion_rate = (status_counts.get("completed", 0) / total_tasks) * 100
result = f"""📊 **Task Analytics**
**Overview:**
• Total Tasks: {total_tasks}
• Completion Rate: {completion_rate:.1f}%
• Overdue Tasks: {overdue_count}
**By Status:**
"""
for status, count in status_counts.items():
percentage = (count / total_tasks) * 100
result += f"• {status.replace('_', ' ').title()}: {count} ({percentage:.1f}%)\n"
result += "\n**By Priority:**\n"
for priority, count in priority_counts.items():
percentage = (count / total_tasks) * 100
result += f"• {priority.title()}: {count} ({percentage:.1f}%)\n"
# Insights
result += "\n**💡 Insights:**\n"
if overdue_count > 0:
result += f"• ⚠️ You have {overdue_count} overdue task(s) that need attention\n"
if priority_counts.get("urgent", 0) > 2:
result += f"• 🔥 High number of urgent tasks - consider prioritizing\n"
if completion_rate < 50:
result += f"• 📈 Low completion rate - try breaking down large tasks\n"
else:
result += f"• 🎉 Good job! You're maintaining a healthy completion rate\n"
return result
except Exception as e:
return f"❌ Error generating analytics: {str(e)}"
# Create the Smart Task Manager Agent
task_manager_agent = Agent(
name="Smart Task Manager",
instructions="""
You are an intelligent task management assistant that helps users organize and track their work efficiently.
# Your Capabilities
- Create new tasks with proper categorization
- List and filter existing tasks
- Update task statuses
- Provide analytics and insights
# Communication Style
- Be proactive in suggesting task organization improvements
- Use emojis and formatting to make responses visually appealing
- Provide actionable insights based on task data
- Ask clarifying questions when task details are incomplete
# Best Practices
- Suggest appropriate priorities based on due dates and descriptions
- Recommend breaking down large tasks into smaller ones
- Alert users about overdue tasks
- Celebrate completed milestones
# When creating tasks:
- Ask about due dates if not provided
- Suggest relevant tags based on task content
- Recommend appropriate priority levels
# When providing analytics:
- Give actionable insights
- Highlight both achievements and areas for improvement
- Suggest specific next actions
""",
tools=[create_task, list_tasks, update_task_status, get_task_analytics]
)
# Demo the Smart Task Manager
async def demo_task_manager():
print("🚀 Smart Task Manager Demo\n")
# Create some sample tasks
queries = [
"Create a task to 'Review Q4 budget proposal' with high priority, due 2025-02-01, tags: finance, budget",
"Add a task 'Buy groceries' with medium priority and tag: personal",
"Create an urgent task 'Fix production bug in payment system' due tomorrow",
"Show me all my tasks",
"Mark the payment system bug as in progress",
"Give me task analytics and insights"
]
for i, query in enumerate(queries, 1):
print(f"**Query {i}:** {query}")
result = await Runner.run(task_manager_agent, query)
print(f"**Response:**\n{result.final_output}\n")
print("-" * 80 + "\n")
if __name__ == "__main__":
import asyncio
asyncio.run(demo_task_manager())
Tổng Kết và Best Practices
Những Điều Cần Ghi Nhớ
✅ Instructions là foundation - Đầu tư thời gian viết instructions chi tiết và rõ ràng
✅ Tools mở rộng capabilities - Function tools cho phép agents tương tác với thế giới thực
✅ Type hints và docstrings - SDK tự động generate schemas từ Python code
✅ Error handling - Luôn handle errors gracefully trong tools
✅ Context awareness - Sử dụng dynamic instructions và context-aware tools
Production Checklist
🔍 Instructions Quality:
- Vai trò và trách nhiệm được định nghĩa rõ ràng
- Phong cách giao tiếp được thiết lập
- Các giới hạn và hạn chế được nêu rõ
- Tiêu chuẩn chất lượng được xác định
🛠️ Tools Development:
- Function signatures có type hints đầy đủ
- Docstrings mô tả parameters và return values
- Error handling toàn diện
- Input validation và sanitization
- Tối ưu hiệu suất cho tools thường dùng
Bước Tiếp Theo
Trong bài tiếp theo, chúng ta sẽ khám phá:
🎯 Context Management - Quản lý data và dependencies hiệu quả
📊 State Management - Duy trì state xuyên suốt conversations
🔄 Data Flow Patterns - Best practices cho data handling
Thử Thách Cho Bạn
Trước khi chuyển bài tiếp theo:
- Extend Task Manager với reminder notifications
- Create domain-specific agent cho work của bạn
- Experiment với different instruction styles và measure effectiveness
Bài tiếp theo: “Context Management: Quản Lý Dữ Liệu và Dependencies Trong Agents” - Chúng ta sẽ học cách quản lý state, data flow, và dependencies để xây dựng agents thực sự thông minh.