Agent 类型深度解析

核心思想：将问题建模为状态图,通过搜索算法找到从初始状态到目标状态的路径。

# 经典的 A* 搜索算法示例
import heapq

def a_star_search(start, goal, get_neighbors, heuristic):
    """
    A* 搜索算法 - Goal-based Agent 的核心

    参数:
        start: 初始状态
        goal: 目标状态
        get_neighbors: 函数,返回当前状态的所有可达邻居状态
        heuristic: 启发式函数,估计从当前状态到目标的代价
    """
    # 优先队列: (f值, 状态, 路径)
    frontier = [(0, start, [])]
    explored = set()

    while frontier:
        f_cost, current, path = heapq.heappop(frontier)

        # 找到目标
        if current == goal:
            return path + [current]

        if current in explored:
            continue
        explored.add(current)

        # 扩展邻居节点
        for neighbor, cost in get_neighbors(current):
            if neighbor not in explored:
                g_cost = len(path) + cost  # 实际代价
                h_cost = heuristic(neighbor, goal)  # 启发式估计
                f_cost = g_cost + h_cost  # 总代价

                heapq.heappush(frontier,
                              (f_cost, neighbor, path + [current]))

    return None  # 无解

# 旅行规划示例
def plan_trip():
    start = "家"
    goal = "完成旅行准备"

    def get_neighbors(state):
        # 定义状态转移
        transitions = {
            "家": [("查机票", 1)],
            "查机票": [("订机票", 2), ("放弃出行", 0)],
            "订机票": [("订酒店", 2)],
            "订酒店": [("规划行程", 1)],
            "规划行程": [("完成旅行准备", 0)]
        }
        return transitions.get(state, [])

    def heuristic(state, goal):
        # 简单的启发式: 估计剩余步骤数
        steps_to_goal = {
            "家": 4,
            "查机票": 3,
            "订机票": 2,
            "订酒店": 1,
            "规划行程": 0,
            "完成旅行准备": 0
        }
        return steps_to_goal.get(state, 999)

    plan = a_star_search(start, goal, get_neighbors, heuristic)
    return plan

# 执行规划
print("旅行规划路径:", plan_trip())
# 输出: ['家', '查机票', '订机票', '订酒店', '规划行程', '完成旅行准备']

核心思想：将复杂任务递归分解为更小的子任务,直到可以直接执行。

class HTNPlanner:
    def __init__(self):
        # 定义任务分解规则
        self.decomposition = {
            "完成旅行": [
                "安排交通",
                "安排住宿",
                "规划行程"
            ],
            "安排交通": [
                "查询航班",
                "比较价格",
                "预订机票",
                "确认订单"
            ],
            "安排住宿": [
                "搜索酒店",
                "筛选位置",
                "预订酒店"
            ],
            "规划行程": [
                "列出景点",
                "优化路线",
                "预订门票"
            ]
        }

        # 定义原子操作(不可再分解)
        self.primitive_tasks = {
            "查询航班", "比较价格", "预订机票", "确认订单",
            "搜索酒店", "筛选位置", "预订酒店",
            "列出景点", "优化路线", "预订门票"
        }

    def plan(self, task, depth=0):
        """递归分解任务"""
        indent = "  " * depth

        # 如果是原子任务,直接执行
        if task in self.primitive_tasks:
            print(f"{indent}✓ 执行: {task}")
            return [task]

        # 否则递归分解
        if task in self.decomposition:
            print(f"{indent}📋 分解任务: {task}")
            plan = []
            for subtask in self.decomposition[task]:
                plan.extend(self.plan(subtask, depth + 1))
            return plan

        return []

# 使用示例
planner = HTNPlanner()
print("=== HTN 规划过程 ===")
final_plan = planner.plan("完成旅行")

print("\n=== 最终执行计划 ===")
for i, action in enumerate(final_plan, 1):
    print(f"{i}. {action}")