跳转至

Hello Agent — 简单智能体

演示 Agent 的核心抽象:感知 → 思考 → 行动循环(Perceive-Think-Act Loop)。 本示例不依赖任何外部 LLM API,使用基于规则的"思考器"驱动,便于在单元测试中验证。 对应教程:第1卷《AI 基础》第1章 什么是 AI Agent。

文件路径: examples/volume-01-ai-fundamentals\hello_agent.py


源代码

"""简单 Agent 示例。

演示 Agent 的核心抽象:**感知 → 思考 → 行动**循环(Perceive-Think-Act Loop)。
本示例不依赖任何外部 LLM API,使用基于规则的"思考器"驱动,便于在单元测试中验证。

对应教程:第1卷《AI 基础》第1章 什么是 AI Agent。
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass, field


@dataclass
class Tool:
    """Agent 可调用的工具。

    Attributes:
        name: 工具名称。
        description: 工具描述,供 Agent 决策时参考。
        func: 工具的实际执行函数,接收字符串参数返回字符串结果。
    """

    name: str
    description: str
    func: Callable[[str], str]

    def run(self, argument: str) -> str:
        """执行工具。

        Args:
            argument: 传给工具的参数字符串。

        Returns:
            工具执行结果字符串。
        """
        return self.func(argument)


@dataclass
class AgentState:
    """Agent 在一次任务中的运行状态。

    Attributes:
        goal: 任务目标。
        steps: 已执行的步骤记录。
        observation: 最近一次感知到的环境信息。
        done: 是否已完成目标。
    """

    goal: str
    steps: list[str] = field(default_factory=list)
    observation: str = ""
    done: bool = False


class HelloAgent:
    """基于规则的简单 Agent。

    通过关键词匹配选择工具,演示 Agent 如何根据观察选择行动。
    真实 Agent 会用 LLM 替换 ``think`` 方法。
    """

    def __init__(self, tools: list[Tool] | None = None, max_steps: int = 5) -> None:
        """初始化 Agent。

        Args:
            tools: 可用工具列表,为空时使用默认工具集。
            max_steps: 最大循环步数,防止无限循环。
        """
        self.tools: dict[str, Tool] = {}
        for tool in tools or self._default_tools():
            self.register_tool(tool)
        self.max_steps = max_steps

    @staticmethod
    def _default_tools() -> list[Tool]:
        return [
            Tool("echo", "原样返回输入内容", lambda arg: f"ECHO: {arg}"),
            Tool("calc", "计算简单算术表达式,如 1+2", HelloAgent._safe_calc),
            Tool("greet", "根据名字生成问候语", lambda name: f"你好,{name}!欢迎使用 AI Agent。"),
        ]

    @staticmethod
    def _safe_calc(expression: str) -> str:
        """安全地计算只含数字与运算符的算术表达式。

        Args:
            expression: 算术表达式字符串。

        Returns:
            计算结果字符串,或在表达式非法时返回错误提示。
        """
        allowed = set("0123456789+-*/(). ")
        if not expression or not set(expression) <= allowed:
            return "错误:仅支持数字与 + - * / ( ) 运算符。"
        try:
            value = eval(expression, {"__builtins__": {}}, {})  # noqa: S307 - 受白名单限制
        except ZeroDivisionError:
            return "错误:除零。"
        except Exception:  # noqa: BLE001
            return "错误:表达式无法计算。"
        return f"{expression} = {value}"

    def register_tool(self, tool: Tool) -> None:
        """注册一个工具。

        Args:
            tool: 待注册的工具实例。
        """
        self.tools[tool.name] = tool

    def think(self, observation: str) -> tuple[str, str]:
        """根据观察决定下一步行动(基于规则的思考器)。

        Args:
            observation: 当前观察到的文本。

        Returns:
            ``(tool_name, argument)`` 元组;若无匹配工具,返回 ``("respond", reply)``。
        """
        text = observation.strip()
        if not text:
            return "respond", "我没有收到任何输入。"
        # 优先匹配算术
        if any(ch.isdigit() for ch in text) and any(op in text for op in "+-*/"):
            return "calc", text
        if "你好" in text or "hello" in text.lower() or "hi" in text.lower():
            return (
                "greet",
                text.replace("你好", "").replace("hello", "").replace("hi", "").strip() or "朋友",
            )
        return "echo", text

    def act(self, tool_name: str, argument: str) -> str:
        """执行选定的行动。

        Args:
            tool_name: 工具名称;``respond`` 表示直接回复。
            argument: 工具参数。

        Returns:
            行动结果字符串。
        """
        if tool_name == "respond":
            return argument
        tool = self.tools.get(tool_name)
        if tool is None:
            return f"未知工具:{tool_name}"
        return tool.run(argument)

    def run(self, goal: str) -> AgentState:
        """执行一次完整的感知-思考-行动循环。

        Args:
            goal: 任务目标/用户输入。

        Returns:
            运行结束后的 ``AgentState``。
        """
        state = AgentState(goal=goal, observation=goal)
        for _ in range(self.max_steps):
            tool_name, argument = self.think(state.observation)
            result = self.act(tool_name, argument)
            state.steps.append(f"{tool_name}({argument!r}) -> {result}")
            state.observation = result
            # 基于规则的 Agent 单步即可完成,标记完成并退出
            state.done = True
            break
        return state


def main() -> None:
    """演示 HelloAgent 的基本用法。"""
    agent = HelloAgent()
    samples = ["你好 Tom", "计算 12 * 8", "echo this please", "3 + 4 * 2"]
    for sample in samples:
        state = agent.run(sample)
        print(f"目标: {sample}")
        for step in state.steps:
            print(f"  步骤: {step}")
        print(f"  最终结果: {state.observation}\n")


if __name__ == "__main__":
    main()