Agent 生命周期

实现一个经典的 ReAct（Reasoning + Acting）Agent，演示完整的生命周期： 感知 → 思考（Thought） → 行动（Action） → 观察（Observation） → 循环 → 完成。 思考过程以文本形式由 LLM 产出，Agent 解析 Thought/Action/Action Input 并执行工具。 LLM 以可注入的回调函数形式提供，便于离线测试。 对应教程：第5卷《Agent 开发》第1章 Agent 生命周期。

文件路径: examples/volume-05-agent\agent_lifecycle.py

---

源代码

"""Agent 生命周期示例。

实现一个经典的 ReAct（Reasoning + Acting）Agent，演示完整的生命周期：
``感知 → 思考（Thought） → 行动（Action） → 观察（Observation） → 循环 → 完成``。

思考过程以文本形式由 LLM 产出，Agent 解析 ``Thought/Action/Action Input`` 并执行工具。
LLM 以可注入的回调函数形式提供，便于离线测试。

对应教程：第5卷《Agent 开发》第1章 Agent 生命周期。
"""

from __future__ import annotations

import json
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from enum import Enum


class LifecycleStage(str, Enum):
    """Agent 生命周期阶段。"""

    INIT = "init"
    PLAN = "plan"
    ACT = "act"
    OBSERVE = "observe"
    DONE = "done"


# LLM 回调类型：接收 prompt 字符串，返回模型输出字符串
LLMCallable = Callable[[str], str]


@dataclass
class AgentStep:
    """一次循环步骤的记录。

    Attributes:
        stage: 所处阶段。
        thought: 思考内容。
        action: 选择的行动/工具名。
        action_input: 行动参数（解析后的字典）。
        observation: 观察结果。
    """

    stage: LifecycleStage
    thought: str = ""
    action: str = ""
    action_input: dict[str, object] = field(default_factory=dict)
    observation: str = ""


@dataclass
class ToolSpec:
    """工具规格说明。

    Attributes:
        name: 工具名。
        description: 描述。
        func: 执行函数，接收关键字参数返回字符串。
    """

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


_THOUGHT_RE = re.compile(r"Thought:\s*(.*)")
_ACTION_RE = re.compile(r"Action:\s*(\S+)")
_ACTION_INPUT_RE = re.compile(r"Action Input:\s*(\{.*\})", re.DOTALL)
_FINAL_ANSWER_RE = re.compile(r"Final Answer:\s*(.*)", re.DOTALL)


def parse_llm_output(text: str) -> tuple[str, str, dict[str, object], str]:
    """解析 LLM 的 ReAct 文本输出。

    Args:
        text: LLM 输出文本。

    Returns:
        ``(thought, action, action_input, final_answer)`` 元组。
        若为最终答案，``action`` 为空、``final_answer`` 非空。

    Raises:
        ValueError: 无法解析 Thought 或既无 Action 也无 Final Answer 时。
    """
    thought_match = _THOUGHT_RE.search(text)
    thought = thought_match.group(1).strip() if thought_match else ""
    # 容错：若整段没有 Thought 标签但含 Final Answer，则 thought 留空
    if not thought and not _FINAL_ANSWER_RE.search(text):
        raise ValueError(f"无法解析 Thought: {text[:80]!r}")

    final_match = _FINAL_ANSWER_RE.search(text)
    if final_match:
        return thought, "", {}, final_match.group(1).strip()

    action_match = _ACTION_RE.search(text)
    input_match = _ACTION_INPUT_RE.search(text)
    if not action_match or not input_match:
        raise ValueError(f"缺少 Action 或 Action Input: {text[:80]!r}")

    action = action_match.group(1).strip()
    try:
        action_input = json.loads(input_match.group(1))
        if not isinstance(action_input, dict):
            raise ValueError("Action Input 必须是 JSON 对象")
    except json.JSONDecodeError as exc:
        raise ValueError(f"Action Input JSON 非法: {input_match.group(1)!r}") from exc

    return thought, action, action_input, ""


class ReActAgent:
    """基于 ReAct 范式的 Agent。

    通过文本化的 Thought/Action/Observation 循环驱动，LLM 可注入。
    """

    def __init__(
        self,
        llm: LLMCallable,
        tools: list[ToolSpec] | None = None,
        max_steps: int = 5,
        system_prompt: str = "你是一个会使用工具的 AI Agent。",
    ) -> None:
        """初始化 Agent。

        Args:
            llm: 可注入的 LLM 回调。
            tools: 可用工具列表。
            max_steps: 最大循环步数。
            system_prompt: 系统提示。
        """
        self.llm = llm
        self.tools: dict[str, ToolSpec] = {t.name: t for t in (tools or [])}
        self.max_steps = max_steps
        self.system_prompt = system_prompt

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

        Args:
            tool: 工具规格。
        """
        self.tools[tool.name] = tool

    def _build_tool_description(self) -> str:
        if not self.tools:
            return "(无可用工具)"
        lines = [f"- {t.name}: {t.description}" for t in self.tools.values()]
        return "\n".join(lines)

    def _build_prompt(self, query: str, history: list[AgentStep]) -> str:
        sections = [
            self.system_prompt,
            f"可用工具：\n{self._build_tool_description()}",
            "请严格按以下格式回复：",
            "Thought: <你的思考>",
            "Action: <工具名>",
            "Action Input: <JSON 参数>",
            "（当你已有答案时，使用：）",
            "Thought: <你的思考>",
            "Final Answer: <最终答案>",
            f"\n用户问题：{query}",
        ]
        if history:
            sections.append("\n历史：")
            for h in history:
                sections.append(
                    f"Thought: {h.thought}\nAction: {h.action}\n"
                    f"Action Input: {json.dumps(h.action_input, ensure_ascii=False)}\n"
                    f"Observation: {h.observation}"
                )
        return "\n\n".join(sections)

    def run(self, query: str) -> tuple[str, list[AgentStep]]:
        """执行一次完整生命周期。

        Args:
            query: 用户问题。

        Returns:
            ``(final_answer, steps)`` 元组。

        Raises:
            RuntimeError: 超过最大步数仍未给出最终答案。
            ValueError: LLM 输出无法解析时。
        """
        history: list[AgentStep] = []
        for _ in range(self.max_steps):
            prompt = self._build_prompt(query, history)
            output = self.llm(prompt)
            thought, action, action_input, final = parse_llm_output(output)

            if final:
                history.append(AgentStep(LifecycleStage.DONE, thought=thought, observation=final))
                return final, history

            tool = self.tools.get(action)
            if tool is None:
                observation = f"错误：未知工具 {action!r}"
            else:
                try:
                    observation = tool.func(**action_input)
                except Exception as exc:  # noqa: BLE001
                    observation = f"错误：工具执行失败 - {exc}"

            history.append(
                AgentStep(
                    LifecycleStage.ACT,
                    thought=thought,
                    action=action,
                    action_input=action_input,
                    observation=observation,
                )
            )

        raise RuntimeError(f"Agent 超过最大步数 {self.max_steps} 仍未结束")


# ===== 内置示例工具 =====


def search_wiki(keyword: str) -> str:
    """示例工具：模拟百科搜索。"""
    db = {
        "AI Agent": "AI Agent 是能感知环境、自主决策并执行任务的智能体。",
        "RAG": "RAG 通过检索外部知识增强大模型生成，缓解幻觉。",
    }
    return db.get(keyword, f"未找到关于 {keyword} 的信息。")


def main() -> None:
    """演示 ReAct Agent 生命周期（使用脚本化 LLM）。"""
    script = [
        "Thought: 我需要先查询 AI Agent 的定义。\n"
        "Action: search_wiki\n"
        'Action Input: {"keyword": "AI Agent"}',
        "Thought: 我已经获得了定义，可以回答了。\n"
        "Final Answer: AI Agent 是能感知环境、自主决策并执行任务的智能体。",
    ]
    iterator = iter(script)

    def fake_llm(_prompt: str) -> str:
        return next(iterator)

    agent = ReActAgent(llm=fake_llm, tools=[ToolSpec("search_wiki", "百科搜索", search_wiki)])
    answer, steps = agent.run("什么是 AI Agent？")
    print(f"最终答案: {answer}\n")
    for i, s in enumerate(steps, 1):
        print(
            f"步骤{i} [{s.stage.value}]: thought={s.thought!r} action={s.action!r} obs={s.observation!r}"
        )


if __name__ == "__main__":
    main()