手搓生产级 AI Agent 系统（3）：Agent 与工具系统完整集成——从 ReAct 到自主决策

┌─────────────────────────────────────────────────┐
│              Agent.run(query)                     │
│                                                   │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐   │
│  │ Thought  │───▶│  Action  │───▶│Observation│  │
│  │ LLM 推理  │    │ 执行工具  │    │ 结果注入  │   │
│  └──────────┘    └──────────┘    └──────────┘   │
│       ▲                                  │        │
│       └──── 检查停止条件 ←──────────────┘        │
│       └──── 未满足 → 继续循环                     │
│       └──── 满足 → 返回最终答案                   │
│                                                   │
│  上下文管理（每步动态组装）：                       │
│  SystemPrompt + 工具列表 + 原始任务 + 历史记录     │
└─────────────────────────────────────────────────┘

from enum import Enum

class StopReason(Enum):
    FINAL_ANSWER = "final_answer"      # LLM 主动输出 Final Answer
    MAX_STEPS = "max_steps"            # 达到最大步数
    MAX_TOKENS = "max_tokens"          # Token 超限
    MAX_COST = "max_cost"              # 费用超限
    NO_PROGRESS = "no_progress"        # 连续 3 步无进展
    DANGEROUS_LOOP = "dangerous_loop"  # 检测到死循环（重复同一工具+参数）

class AgentStopper:
    """四层停止条件检查器"""

    def __init__(self, max_steps: int = 15, max_tokens: int = 50000,
                 max_cost: float = 0.50, no_progress_threshold: int = 3):
        self.max_steps = max_steps
        self.max_tokens = max_tokens
        self.max_cost = max_cost
        self.no_progress_threshold = no_progress_threshold
        self.total_tokens = 0
        self.total_cost = 0.0
        self.steps = 0
        self.last_actions = []      # 最近 N 个 action 记录
        self.last_useful_result = 0  # 最后拿到有用结果的步数

    def check(self, step: int, tokens_this_step: int, cost_this_step: float,
              action: str = None, params: dict = None,
              result: str = None, has_final: bool = False) -> tuple[bool, StopReason]:
        self.steps = step
        self.total_tokens += tokens_this_step
        self.total_cost += cost_this_step

        # Layer 1: LLM 主动结束
        if has_final:
            return True, StopReason.FINAL_ANSWER

        # Layer 2: 硬上限
        if step >= self.max_steps:
            return True, StopReason.MAX_STEPS
        if self.total_tokens >= self.max_tokens:
            return True, StopReason.MAX_TOKENS
        if self.total_cost >= self.max_cost:
            return True, StopReason.MAX_COST

        # Layer 3: 连续无效步数检测
        if action and result:
            action_key = f"{action}:{json.dumps(params, sort_keys=True)}"
            self.last_actions.append(action_key)

            # 检查最近 no_progress_threshold 个 action 是否都相同
            if len(self.last_actions) > self.no_progress_threshold:
                self.last_actions.pop(0)
            if (len(self.last_actions) >= self.no_progress_threshold and
                len(set(self.last_actions[-self.no_progress_threshold:])) == 1):
                return True, StopReason.DANGEROUS_LOOP

            # 检查是否有实质性进展
            if result and "错误" not in result and "查不到" not in result:
                self.last_useful_result = step

        # Layer 4: 长期无进展
        if step - self.last_useful_result > self.no_progress_threshold + 2:
            return True, StopReason.NO_PROGRESS

        return False, None

class ContextBuilder:
    """每步动态构建 LLM 上下文"""

    def __init__(self, system_prompt: str, tools_desc: str):
        self.system_prompt = system_prompt
        self.tools_desc = tools_desc
        self.original_query = ""      # 原始问题，每步注入防漂移
        self.steps: list[dict] = []   # 每步的 Thought/Action/Observation

    def init(self, user_query: str):
        self.original_query = user_query
        self.steps = []

    def add_step(self, thought: str, action: str, observation: str,
                 tokens: int, cost: float):
        self.steps.append({
            "thought": thought,
            "action": action,
            "observation": observation[:800],  # 截断长结果
            "tokens": tokens,
            "cost": cost,
        })

    def build(self) -> list[dict]:
        """构建给 LLM 的完整消息列表"""
        # 任务摘要（每步注入，防漂移）
        task_reminder = f"原始任务：{self.original_query[:200]}"
        if len(self.steps) > 0:
            task_reminder += f"\n当前进度：已完成 {len(self.steps)} 步"

        messages = [
            {"role": "system", "content": self.system_prompt.format(
                tools_description=self.tools_desc
            )},
            {"role": "system", "content": task_reminder},
        ]

        # 注入最近 5 步的完整记录，更早的只保留摘要
        recent = self.steps[-5:]
        older = self.steps[:-5]

        if older:
            summary = self._summarize_older_steps(older)
            messages.append({"role": "system", "content": f"之前的执行摘要：{summary}"})

        for s in recent:
            messages.append({
                "role": "assistant",
                "content": f"Thought: {s['thought']}\nAction: {s['action']}"
            })
            messages.append({
                "role": "user",
                "content": f"Observation: {s['observation']}"
            })

        return messages

    def _summarize_older_steps(self, steps: list) -> str:
        """压缩早期步骤——只保留关键信息，减少 Token 消耗"""
        lines = []
        for i, s in enumerate(steps):
            action_short = s["action"][:60]
            result_short = s["observation"][:100]
            lines.append(f"Step{i+1}: {action_short} → {result_short}")
        return "; ".join(lines)

    @property
    def total_cost(self) -> float:
        return sum(s["cost"] for s in self.steps)

    @property
    def total_tokens(self) -> int:
        return sum(s["tokens"] for s in self.steps)

class IntegratedAgent:
    """完整的 ReAct Agent——循环 + 工具 + 上下文 + 停止条件"""

    def __init__(self, tools: dict, llm_client, model: str = "deepseek-chat"):
        self.tools = tools
        self.llm = llm_client
        self.model = model

        self.tools_desc = self._build_tools_description()

        self.system_prompt = """你是一个自主 AI Agent。严格按照 ReAct 格式回复。

## 可用工具
{tools_description}

## 回复格式
需要调用工具时：
Thought: 
Action: 
Action Input: 

已获得足够信息时：
Thought: 
Final Answer: 

## 规则
1. 每次只能调用一个工具
2. 工具返回后，先思考再决定下一步
3. 连续 2 次返回错误，直接告知用户

原始任务：系统会自动在每条消息中注入"""

        self.context = ContextBuilder(self.system_prompt, self.tools_desc)
        self.stopper = AgentStopper()
        self.executor = SecureToolExecutor()
        self.validator = ParameterValidator()

    def run(self, query: str) -> dict:
        """运行 Agent，返回完整执行轨迹"""
        self.context.init(query)
        trace = []

        for step in range(1, self.stopper.max_steps + 1):
            # 1. 构建上下文
            messages = self.context.build()

            # 2. 调用 LLM
            response = self.llm.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0.1,
            )
            llm_output = response.choices[0].message.content
            tokens_this_step = response.usage.total_tokens
            cost_this_step = tokens_this_step * 2 / 1_000_000  # ¥2/1M

            print(f"\n{'='*60}")
            print(f"Step {step} | Token: {tokens_this_step} | Cost: ¥{cost_this_step:.4f}")
            print(f"LLM: {llm_output[:200]}")

            # 3. 解析输出
            parsed = self._parse_output(llm_output)

            if parsed["type"] == "final":
                self.context.add_step(parsed["thought"], "Final Answer",
                                     parsed["answer"], tokens_this_step, cost_this_step)
                trace.append({"step": step, "type": "final", "answer": parsed["answer"]})
                return {
                    "success": True,
                    "answer": parsed["answer"],
                    "steps": step,
                    "total_cost": self.context.total_cost + cost_this_step,
                    "total_tokens": self.context.total_tokens + tokens_this_step,
                    "trace": trace,
                    "stop_reason": "final_answer",
                }

            if parsed["type"] == "action":
                # 4. 执行工具
                tool_name = parsed["tool"]
                tool_input = parsed["input"]

                if tool_name not in self.tools:
                    observation = f"错误：工具 '{tool_name}' 不存在。可用: {list(self.tools.keys())}"
                else:
                    tool = self.tools[tool_name]
                    valid, err = self.validator.validate(tool, {"input": tool_input})
                    if not valid:
                        observation = f"参数错误: {err}"
                    else:
                        exec_result = self.executor.execute(tool, {"input": tool_input})
                        observation = (exec_result.get("result") or
                                      exec_result.get("error", "未知错误"))

                print(f"🔧 {tool_name}({str(tool_input)[:50]})")
                print(f"📋 {str(observation)[:100]}")

                self.context.add_step(parsed["thought"],
                                     f"{tool_name}({tool_input})",
                                     str(observation),
                                     tokens_this_step, cost_this_step)
                trace.append({"step": step, "type": "action",
                             "tool": tool_name, "input": tool_input,
                             "observation": str(observation)[:200]})

                # 5. 检查停止条件
                should_stop, reason = self.stopper.check(
                    step, tokens_this_step, cost_this_step,
                    action=tool_name, params={"input": str(tool_input)},
                    result=str(observation)
                )

                if should_stop:
                    # 达到硬上限，强制让 LLM 总结
                    if reason != StopReason.FINAL_ANSWER:
                        summary = self._force_final_answer()
                        return {
                            "success": True,
                            "answer": summary,
                            "steps": step,
                            "total_cost": self.context.total_cost,
                            "total_tokens": self.context.total_tokens,
                            "trace": trace,
                            "stop_reason": reason.value,
                        }

        # 兜底
        return {"success": False, "error": "超出最大步数", "trace": trace}

    def _parse_output(self, text: str) -> dict:
        """解析 LLM 输出"""
        # Final Answer 检测
        import re
        fa = re.search(r'Final Answer:\s*([\s\S]+?)$', text, re.IGNORECASE)
        if fa:
            thought = text[:fa.start()].strip()
            return {"type": "final", "thought": thought, "answer": fa.group(1).strip()}

        # Action 检测
        act = re.search(r'Action:\s*(\w+)\s*\nAction Input:\s*(.*?)(?:\n|$)',
                       text, re.IGNORECASE)
        if act:
            thought = text[:act.start()].strip()
            return {"type": "action", "thought": thought,
                   "tool": act.group(1).strip(), "input": act.group(2).strip()}

        # 无法解析——让 LLM 再试
        return {"type": "unparseable", "raw": text}

    def _force_final_answer(self) -> str:
        """强制 LLM 基于已有信息给出答案"""
        messages = self.context.build()
        messages.append({
            "role": "user",
            "content": "已达到最大步数。请基于已有的所有信息，给出你的 Final Answer。不要调用工具。"
        })
        response = self.llm.chat.completions.create(
            model=self.model, messages=messages, temperature=0.1
        )
        return response.choices[0].message.content

    def _build_tools_description(self) -> str:
        lines = []
        for name, tool in self.tools.items():
            params = ", ".join(
                f"{p.name}: {p.type}" + ("?" if not p.required else "")
                for p in tool.parameters
            )
            lines.append(f"- {name}({params}): {tool.description}")
        return "\n".join(lines)