本文基于开源项目进行解读与拓展

• 官方项目地址：GitHub - microsoft/ai-agents-for-beginners
• 项目许可证：MIT License

若想跑通代码建议去官网查看，个人觉得学习其思路也是获益良多，每个流程间如何衔接。

07-planning-design

定义总体目标并分解任务

大多数实际任务都太复杂了，无法一步完成。AI 代理需要一个简洁的目标来指导其规划和行动。例如，考虑目标：

"Generate a 3-day travel itinerary."

虽然说起来很简单，但它仍然需要改进。目标越明确，代理人（和任何人类合作者）就越能专注于实现正确的结果，例如创建包含航班选择、酒店推荐和活动建议的综合行程。

1.任务分解

大型或复杂的任务在拆分为较小的、面向目标的子任务时，将变得更加易于管理。对于旅行路线示例，您可以将目标分解为：

• Flight Booking 航班预订
• Hotel Booking 酒店预订
• Car Rental 租车
• Personalization 个性化

然后，每个子任务都可以由专门的代理或流程处理。一个代理可能专门搜索最佳航班优惠，另一个代理专注于酒店预订，依此类推。然后，协调代理或“下游”代理可以将这些结果编译成一个供最终用户使用的有凝聚力的路线。

这种模块化方法还允许增量增强。例如，可以为 Food Recommendations 或 Local Activity Suggestions 添加专门的代理，并随着时间的推移优化行程。

2.结构化输出

大型语言模型 （LLM） 可以生成结构化输出 （e.g. 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://docs.github.com/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))

3.使用多代理编排规划代理

在此示例中，语义路由器代理 （Semantic Router Agent ） 收到用户请求（例如，“我需要为我的旅行提供酒店计划”。）

• 接收酒店计划：规划者接收用户的消息，并根据系统提示（包括可用的代理详细信息）生成结构化的旅行计划。
• 列出代理及其工具：代理注册表包含代理列表（例如，航班、酒店、汽车租赁和活动）以及他们提供的功能或工具。
• 将计划路由到相应的代理：根据子任务的数量，规划者要么将消息直接发送到专用代理（用于单任务场景），要么通过群聊管理器进行协调以进行多代理协作。
• 总结结果：最后，规划者总结生成的计划以清晰起见。以下 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."
        }
    ]
}

4.迭代规划

有些任务需要来回或重新规划，其中一个子任务的结果会影响下一个子任务。例如，如果代理在预订航班时发现意外的数据格式，则可能需要在继续预订酒店之前调整其策略。

此外，用户反馈（例如，员工决定他们更喜欢早点的航班）可能会触发部分重新计划。这种动态的迭代方法可确保最终解决方案与实际约束和不断变化的用户偏好保持一致。

例如示例代码：

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