ai-agents-for-beginners
（点击上方图片观看本课程视频）
规划设计
引言
本课程将涵盖
- 定义清晰的总体目标，并将复杂的任务分解为可管理的任务。
- 利用结构化输出以获得更可靠且机器可读的响应。
- 采用事件驱动方法来处理动态任务和意外输入。
学习目标
完成本课程后，您将了解以下内容：
- 识别并为 AI 代理设置总体目标，确保它清楚地知道需要实现什么。
- 将复杂的任务分解为可管理的子任务，并将其组织成逻辑序列。
- 为代理配备正确的工具（例如，搜索工具或数据分析工具），决定何时以及如何使用它们，并处理出现的意外情况。
- 评估子任务结果，衡量性能，并迭代行动以改进最终输出。
定义总体目标并分解任务
大多数实际任务过于复杂，无法一步完成。AI 代理需要一个简洁的目标来指导其规划和行动。例如，考虑目标：
"Generate a 3-day travel itinerary."
虽然表述简单，但仍需完善。目标越清晰，代理（以及任何人类协作方）就越能专注于实现正确的结果，例如创建包含航班选项、酒店推荐和活动建议的全面行程。
任务分解
当大型或复杂的任务分解为更小的、面向目标的子任务时，它们会变得更易于管理。对于旅行行程示例，您可以将目标分解为：
- 航班预订
- 酒店预订
- 汽车租赁
- 个性化
然后，每个子任务都可以由专门的代理或流程处理。一个代理可能专门搜索最佳航班交易，另一个专注于酒店预订，依此类推。然后，协调或“下游”代理可以将这些结果编译成一个连贯的行程呈现给最终用户。
这种模块化方法也允许增量增强。例如，您可以添加用于餐饮推荐或本地活动建议的专门代理，并随着时间的推移完善行程。
结构化输出
大型语言模型 (LLM) 可以生成结构化输出（例如 JSON），这更易于下游代理或服务解析和处理。这在多代理环境中特别有用，我们可以在接收到规划输出后执行这些任务。请参阅这篇博客文章以快速了解。
以下 Python 片段演示了一个简单的规划代理将目标分解为子任务并生成结构化计划：
from pydantic import BaseModel
from enum import Enum
from typing import List, Optional, Union
import json
import os
from typing import Optional
from pprint import pprint
from autogen_core.models import UserMessage, SystemMessage, AssistantMessage
from autogen_ext.models.azure import AzureAIChatCompletionClient
from azure.core.credentials import AzureKeyCredential
class AgentEnum(str, Enum):
FlightBooking = "flight_booking"
HotelBooking = "hotel_booking"
CarRental = "car_rental"
ActivitiesBooking = "activities_booking"
DestinationInfo = "destination_info"
DefaultAgent = "default_agent"
GroupChatManager = "group_chat_manager"
# Travel SubTask Model
class TravelSubTask(BaseModel):
task_details: str
assigned_agent: AgentEnum # we want to assign the task to the agent
class TravelPlan(BaseModel):
main_task: str
subtasks: List[TravelSubTask]
is_greeting: bool
client = AzureAIChatCompletionClient(
model="gpt-4o-mini",
endpoint="https://models.inference.ai.azure.com",
# To authenticate with the model you will need to generate a personal access token (PAT) in your GitHub settings.
# Create your PAT token by following instructions here: https://githubdocs.cn/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens
credential=AzureKeyCredential(os.environ["GITHUB_TOKEN"]),
model_info={
"json_output": False,
"function_calling": True,
"vision": True,
"family": "unknown",
},
)
# Define the user message
messages = [
SystemMessage(content="""You are an planner agent.
Your job is to decide which agents to run based on the user's request.
Provide your response in JSON format with the following structure:
{'main_task': 'Plan a family trip from Singapore to Melbourne.',
'subtasks': [{'assigned_agent': 'flight_booking',
'task_details': 'Book round-trip flights from Singapore to '
'Melbourne.'}
Below are the available agents specialised in different tasks:
- FlightBooking: For booking flights and providing flight information
- HotelBooking: For booking hotels and providing hotel information
- CarRental: For booking cars and providing car rental information
- ActivitiesBooking: For booking activities and providing activity information
- DestinationInfo: For providing information about destinations
- DefaultAgent: For handling general requests""", source="system"),
UserMessage(
content="Create a travel plan for a family of 2 kids from Singapore to Melboune", source="user"),
]
response = await client.create(messages=messages, extra_create_args={"response_format": 'json_object'})
response_content: Optional[str] = response.content if isinstance(
response.content, str) else None
if response_content is None:
raise ValueError("Response content is not a valid JSON string" )
pprint(json.loads(response_content))
# # Ensure the response content is a valid JSON string before loading it
# response_content: Optional[str] = response.content if isinstance(
# response.content, str) else None
# if response_content is None:
# raise ValueError("Response content is not a valid JSON string")
# # Print the response content after loading it as JSON
# pprint(json.loads(response_content))
# Validate the response content with the MathReasoning model
# TravelPlan.model_validate(json.loads(response_content))
具有多代理编排的规划代理
在此示例中，语义路由代理接收用户请求（例如，“我需要我的旅行酒店计划。”）。
然后规划器执行以下操作：
- 接收酒店计划：规划器接收用户的消息，并根据系统提示（包括可用的代理详细信息）生成结构化的旅行计划。
- 列出代理及其工具：代理注册表包含一个代理列表（例如，用于航班、酒店、汽车租赁和活动），以及它们提供的功能或工具。
- 将计划路由到相应的代理：根据子任务的数量，规划器要么将消息直接发送到专用代理（用于单任务场景），要么通过群聊管理器进行协调以进行多代理协作。
- 总结结果：最后，规划器总结生成的计划以提高清晰度。以下 Python 代码示例说明了这些步骤：
from pydantic import BaseModel
from enum import Enum
from typing import List, Optional, Union
class AgentEnum(str, Enum):
FlightBooking = "flight_booking"
HotelBooking = "hotel_booking"
CarRental = "car_rental"
ActivitiesBooking = "activities_booking"
DestinationInfo = "destination_info"
DefaultAgent = "default_agent"
GroupChatManager = "group_chat_manager"
# Travel SubTask Model
class TravelSubTask(BaseModel):
task_details: str
assigned_agent: AgentEnum # we want to assign the task to the agent
class TravelPlan(BaseModel):
main_task: str
subtasks: List[TravelSubTask]
is_greeting: bool
import json
import os
from typing import Optional
from autogen_core.models import UserMessage, SystemMessage, AssistantMessage
from autogen_ext.models.openai import AzureOpenAIChatCompletionClient
# Create the client with type-checked environment variables
client = AzureOpenAIChatCompletionClient(
azure_deployment=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),
model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),
api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
)
from pprint import pprint
# Define the user message
messages = [
SystemMessage(content="""You are an planner agent.
Your job is to decide which agents to run based on the user's request.
Below are the available agents specialized in different tasks:
- FlightBooking: For booking flights and providing flight information
- HotelBooking: For booking hotels and providing hotel information
- CarRental: For booking cars and providing car rental information
- ActivitiesBooking: For booking activities and providing activity information
- DestinationInfo: For providing information about destinations
- DefaultAgent: For handling general requests""", source="system"),
UserMessage(content="Create a travel plan for a family of 2 kids from Singapore to Melbourne", source="user"),
]
response = await client.create(messages=messages, extra_create_args={"response_format": TravelPlan})
# Ensure the response content is a valid JSON string before loading it
response_content: Optional[str] = response.content if isinstance(response.content, str) else None
if response_content is None:
raise ValueError("Response content is not a valid JSON string")
# Print the response content after loading it as JSON
pprint(json.loads(response_content))
以下是前面代码的输出，然后您可以使用此结构化输出路由到 assigned_agent 并向最终用户总结旅行计划。
{
"is_greeting": "False",
"main_task": "Plan a family trip from Singapore to Melbourne.",
"subtasks": [
{
"assigned_agent": "flight_booking",
"task_details": "Book round-trip flights from Singapore to Melbourne."
},
{
"assigned_agent": "hotel_booking",
"task_details": "Find family-friendly hotels in Melbourne."
},
{
"assigned_agent": "car_rental",
"task_details": "Arrange a car rental suitable for a family of four in Melbourne."
},
{
"assigned_agent": "activities_booking",
"task_details": "List family-friendly activities in Melbourne."
},
{
"assigned_agent": "destination_info",
"task_details": "Provide information about Melbourne as a travel destination."
}
]
}
包含前面代码示例的示例笔记本可在此处获取。
迭代规划
有些任务需要反复或重新规划，其中一个子任务的结果会影响下一个子任务。例如，如果代理在预订航班时发现意外的数据格式，它可能需要调整其策略，然后才能继续预订酒店。
此外，用户反馈（例如人类决定他们更喜欢更早的航班）可以触发部分重新规划。这种动态的迭代方法确保最终解决方案符合现实世界的约束和不断变化的用户偏好。
例如，示例代码
from autogen_core.models import UserMessage, SystemMessage, AssistantMessage
#.. same as previous code and pass on the user history, current plan
messages = [
SystemMessage(content="""You are a planner agent to optimize the
Your job is to decide which agents to run based on the user's request.
Below are the available agents specialized in different tasks:
- FlightBooking: For booking flights and providing flight information
- HotelBooking: For booking hotels and providing hotel information
- CarRental: For booking cars and providing car rental information
- ActivitiesBooking: For booking activities and providing activity information
- DestinationInfo: For providing information about destinations
- DefaultAgent: For handling general requests""", source="system"),
UserMessage(content="Create a travel plan for a family of 2 kids from Singapore to Melbourne", source="user"),
AssistantMessage(content=f"Previous travel plan - {TravelPlan}", source="assistant")
]
# .. re-plan and send the tasks to respective agents
如需更全面的规划，请查看 Magnetic One 博客文章，了解如何解决复杂任务。
总结
在本文中，我们看了一个示例，说明如何创建可以动态选择已定义的可用代理的规划器。规划器的输出分解任务并分配代理，以便它们可以执行。假设代理可以访问执行任务所需的功能/工具。除了代理之外，您还可以包含其他模式，如反射、摘要器和轮询聊天，以进一步自定义。
额外资源
AutoGen Magentic One - 一个用于解决复杂任务的通用多代理系统，并在多个具有挑战性的代理基准测试中取得了令人印象深刻的结果。参考：autogen-magentic-one。在此实现中，协调器创建特定于任务的计划并将这些任务委派给可用的代理。除了规划之外，协调器还采用跟踪机制来监控任务进度并根据需要重新规划。
对规划设计模式还有更多疑问吗？
加入 Azure AI Foundry Discord，与其他学习者交流，参加办公时间，并获得您的 AI Agent 问题解答。