AI 论文解读：Agentic Reasoning 实现指南 2026

"""
ReAct 模式的核心循环实现。

LLM 用自然语言进行推理、调用工具、观察结果，
这一循环最多重复 max_steps 次。
"""
from dataclasses import dataclass, field
from typing import Callable, Optional
from enum import Enum
import json
import re


class StepType(Enum):
    THOUGHT = "thought"
    ACTION = "action"
    OBSERVATION = "observation"
    FINAL_ANSWER = "final_answer"


@dataclass
class AgentStep:
    step_type: StepType
    content: str
    tool_name: Optional[str] = None
    tool_input: Optional[dict] = None
    token_count: int = 0


@dataclass
class AgentTrace:
    """智能体执行的完整记录。"""
    question: str
    steps: list[AgentStep] = field(default_factory=list)
    final_answer: Optional[str] = None
    total_tokens: int = 0
    total_tool_calls: int = 0

    def add_step(self, step: AgentStep):
        self.steps.append(step)
        self.total_tokens += step.token_count
        if step.step_type == StepType.ACTION:
            self.total_tool_calls += 1


class ReActAgent:
    """ReAct 模式智能体。

    接收 LLM 和一组工具，针对给定问题
    反复执行推理-行动-观察循环。
    """

    SYSTEM_PROMPT = """You are a helpful assistant that solves problems step by step.
For each step, you MUST output exactly one of:
- Thought: <your reasoning about what to do next>
- Action: <tool_name>({"param": "value"})
- Final Answer: <your final response to the user>

Available tools:
{tool_descriptions}

Rules:
- Always think before acting.
- After observing a tool result, think about what it means before the next action.
- When you have enough information, provide Final Answer.
"""

    def __init__(
        self,
        llm: Callable,   # (messages: list[dict]) -> str
        tools: dict[str, Callable],
        tool_descriptions: dict[str, str],
        max_steps: int = 10,
        max_tokens_per_step: int = 1024,
    ):
        self.llm = llm
        self.tools = tools
        self.tool_descriptions = tool_descriptions
        self.max_steps = max_steps
        self.max_tokens_per_step = max_tokens_per_step

    def run(self, question: str) -> AgentTrace:
        trace = AgentTrace(question=question)

        # 将工具说明插入系统提示词
        tool_desc_text = "\n".join(
            f"- {name}: {desc}"
            for name, desc in self.tool_descriptions.items()
        )
        system_msg = self.SYSTEM_PROMPT.format(tool_descriptions=tool_desc_text)

        messages = [
            {"role": "system", "content": system_msg},
            {"role": "user", "content": question},
        ]

        for step_num in range(self.max_steps):
            # 请求 LLM 生成下一步
            response = self.llm(messages)
            parsed = self._parse_response(response)

            if parsed.step_type == StepType.FINAL_ANSWER:
                trace.final_answer = parsed.content
                trace.add_step(parsed)
                break

            trace.add_step(parsed)
            messages.append({"role": "assistant", "content": response})

            if parsed.step_type == StepType.ACTION and parsed.tool_name:
                # 执行工具
                observation = self._execute_tool(
                    parsed.tool_name, parsed.tool_input or {}
                )
                obs_step = AgentStep(
                    step_type=StepType.OBSERVATION,
                    content=observation,
                )
                trace.add_step(obs_step)
                messages.append({
                    "role": "user",
                    "content": f"Observation: {observation}",
                })

        return trace

    def _parse_response(self, response: str) -> AgentStep:
        """解析 LLM 输出，区分 Thought/Action/Final Answer。"""
        response = response.strip()

        # 检查 Final Answer
        if response.lower().startswith("final answer:"):
            return AgentStep(
                step_type=StepType.FINAL_ANSWER,
                content=response[len("final answer:"):].strip(),
            )

        # 解析 Action：Action: tool_name({"key": "value"})
        action_match = re.match(
            r'Action:\s*(\w+)\((\{.*\})\)', response, re.DOTALL
        )
        if action_match:
            tool_name = action_match.group(1)
            try:
                tool_input = json.loads(action_match.group(2))
            except json.JSONDecodeError:
                tool_input = {}
            return AgentStep(
                step_type=StepType.ACTION,
                content=response,
                tool_name=tool_name,
                tool_input=tool_input,
            )

        # 其余情况按 Thought 处理
        return AgentStep(
            step_type=StepType.THOUGHT,
            content=response,
        )

    def _execute_tool(self, tool_name: str, tool_input: dict) -> str:
        """执行工具并将结果以字符串形式返回。"""
        if tool_name not in self.tools:
            return f"Error: Unknown tool '{tool_name}'. Available: {list(self.tools.keys())}"
        try:
            result = self.tools[tool_name](**tool_input)
            return str(result)
        except Exception as e:
            return f"Error executing {tool_name}: {type(e).__name__}: {str(e)}"

"""
生产环境中的智能体工具定义。

每个工具都内置输入校验、timeout、成本限制，
并以结构化形式返回执行结果。
"""
from dataclasses import dataclass
from typing import Any, Optional
import httpx
import time


@dataclass
class ToolResult:
    success: bool
    data: Any
    error: Optional[str] = None
    execution_time_ms: float = 0.0
    cost_usd: float = 0.0


class WebSearchTool:
    """网页搜索工具。

    在智能体需要查询最新信息时使用。
    内置 rate limit 与成本限制。
    """

    def __init__(
        self,
        api_key: str,
        max_results: int = 5,
        timeout_seconds: float = 10.0,
        max_calls_per_minute: int = 10,
    ):
        self.api_key = api_key
        self.max_results = max_results
        self.timeout_seconds = timeout_seconds
        self.max_calls_per_minute = max_calls_per_minute
        self._call_timestamps: list[float] = []

    def _check_rate_limit(self) -> bool:
        now = time.time()
        self._call_timestamps = [
            ts for ts in self._call_timestamps if now - ts < 60
        ]
        return len(self._call_timestamps) < self.max_calls_per_minute

    def __call__(self, query: str) -> ToolResult:
        if not query or len(query) > 500:
            return ToolResult(
                success=False,
                data=None,
                error="Query must be 1-500 characters",
            )

        if not self._check_rate_limit():
            return ToolResult(
                success=False,
                data=None,
                error=f"Rate limit exceeded: max {self.max_calls_per_minute}/min",
            )

        start = time.monotonic()
        try:
            # 实际的搜索 API 调用（例如 Tavily、Serper 等）
            with httpx.Client(timeout=self.timeout_seconds) as client:
                response = client.get(
                    "https://api.search-provider.com/search",
                    params={"q": query, "max_results": self.max_results},
                    headers={"Authorization": f"Bearer {self.api_key}"},
                )
                response.raise_for_status()
                elapsed = (time.monotonic() - start) * 1000
                self._call_timestamps.append(time.time())

                return ToolResult(
                    success=True,
                    data=response.json(),
                    execution_time_ms=elapsed,
                    cost_usd=0.001,  # 每次调用的预估成本
                )
        except httpx.TimeoutException:
            return ToolResult(
                success=False,
                data=None,
                error=f"Search timed out after {self.timeout_seconds}s",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )
        except httpx.HTTPStatusError as e:
            return ToolResult(
                success=False,
                data=None,
                error=f"HTTP {e.response.status_code}: {e.response.text[:200]}",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )


class CodeExecutionTool:
    """代码执行工具。

    智能体通过执行 Python 代码来完成计算或数据处理。
    出于安全考虑，只允许 import 白名单内的模块，并限制执行时间和内存。
    """

    ALLOWED_MODULES = {"math", "statistics", "json", "re", "datetime", "collections"}

    def __init__(self, timeout_seconds: float = 5.0):
        self.timeout_seconds = timeout_seconds

    def __call__(self, code: str) -> ToolResult:
        if not code or len(code) > 5000:
            return ToolResult(
                success=False,
                data=None,
                error="Code must be 1-5000 characters",
            )

        # import 检查：只允许使用白名单内的模块
        import_lines = [
            line.strip() for line in code.splitlines()
            if line.strip().startswith("import ") or line.strip().startswith("from ")
        ]
        for line in import_lines:
            module = line.split()[1].split(".")[0]
            if module not in self.ALLOWED_MODULES:
                return ToolResult(
                    success=False,
                    data=None,
                    error=f"Module '{module}' not allowed. Allowed: {self.ALLOWED_MODULES}",
                )

        start = time.monotonic()
        try:
            # 在受限环境中执行
            local_vars: dict = {}
            exec(code, {"__builtins__": {}}, local_vars)  # noqa: S102

            elapsed = (time.monotonic() - start) * 1000
            # 若存在 'result' 变量，则返回该变量
            result = local_vars.get("result", str(local_vars))

            return ToolResult(
                success=True,
                data=result,
                execution_time_ms=elapsed,
            )
        except Exception as e:
            return ToolResult(
                success=False,
                data=None,
                error=f"{type(e).__name__}: {str(e)}",
                execution_time_ms=(time.monotonic() - start) * 1000,
            )

"""
智能体的工作记忆（working memory）管理。

保留完整的对话历史，但在传给 LLM 时，
会基于重要度进行摘要/筛选，以适配上下文窗口。
"""
from dataclasses import dataclass, field
from typing import Optional
import hashlib


@dataclass
class MemoryEntry:
    role: str
    content: str
    step_number: int
    importance: float = 0.5  # 0.0 ~ 1.0
    token_count: int = 0
    content_hash: str = ""

    def __post_init__(self):
        if not self.content_hash:
            self.content_hash = hashlib.md5(
                self.content.encode()
            ).hexdigest()[:8]


class SlidingWindowMemory:
    """滑动窗口 + 基于重要度的记忆管理。

    始终保留最近 K 条消息，
    更早的消息则按 importance 分数筛选。
    """

    def __init__(
        self,
        max_tokens: int = 8192,
        recent_window: int = 6,     # 始终保留最近 N 条
        system_prompt_tokens: int = 500,
    ):
        self.max_tokens = max_tokens
        self.recent_window = recent_window
        self.system_prompt_tokens = system_prompt_tokens
        self.entries: list[MemoryEntry] = []

    def add(self, entry: MemoryEntry):
        # 防止重复
        if any(e.content_hash == entry.content_hash for e in self.entries):
            return
        self.entries.append(entry)

    def get_context(self, system_message: str) -> list[dict]:
        """构建要传给 LLM 的消息列表。

        1. 始终包含系统提示词
        2. 始终包含最近 recent_window 条
        3. 其余按 importance 顺序在预算内纳入
        """
        budget = self.max_tokens - self.system_prompt_tokens
        messages = [{"role": "system", "content": system_message}]

        if not self.entries:
            return messages

        # 先确保最近的消息
        recent = self.entries[-self.recent_window:]
        older = self.entries[:-self.recent_window] if len(self.entries) > self.recent_window else []

        recent_tokens = sum(e.token_count for e in recent)

        # 在预算内加入较旧消息中重要度高的部分
        remaining_budget = budget - recent_tokens
        selected_older = sorted(older, key=lambda e: e.importance, reverse=True)

        included_older = []
        for entry in selected_older:
            if remaining_budget <= 0:
                break
            if entry.token_count <= remaining_budget:
                included_older.append(entry)
                remaining_budget -= entry.token_count

        # 按时间顺序排列，构成消息列表
        included_older.sort(key=lambda e: e.step_number)
        all_entries = included_older + recent

        for entry in all_entries:
            messages.append({"role": entry.role, "content": entry.content})

        return messages

    def mark_important(self, step_number: int, importance: float = 1.0):
        """提高特定步骤的重要度。

        用于标记工具执行结果、关键发现等。
        """
        for entry in self.entries:
            if entry.step_number == step_number:
                entry.importance = importance
                break

"""
用于控制智能体执行的成本与资源使用的护栏（guardrail）。
"""
from dataclasses import dataclass
from typing import Optional
import time


@dataclass
class AgentBudget:
    max_llm_calls: int = 15
    max_tool_calls: int = 10
    max_total_tokens: int = 50_000
    max_cost_usd: float = 0.50
    max_wall_time_seconds: float = 120.0


@dataclass
class AgentUsage:
    llm_calls: int = 0
    tool_calls: int = 0
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    start_time: float = 0.0

    def elapsed_seconds(self) -> float:
        return time.time() - self.start_time if self.start_time else 0.0


class BudgetGuard:
    """智能体执行预算监视器。

    每一步执行前调用 check()，确认是否超出预算。
    一旦超出，智能体应以当前已有的结果提前终止。
    """

    def __init__(self, budget: AgentBudget):
        self.budget = budget
        self.usage = AgentUsage()

    def start(self):
        self.usage.start_time = time.time()

    def record_llm_call(self, tokens: int, cost_usd: float):
        self.usage.llm_calls += 1
        self.usage.total_tokens += tokens
        self.usage.total_cost_usd += cost_usd

    def record_tool_call(self, cost_usd: float = 0.0):
        self.usage.tool_calls += 1
        self.usage.total_cost_usd += cost_usd

    def check(self) -> Optional[str]:
        """预算超出时返回原因，正常则返回 None。"""
        if self.usage.llm_calls >= self.budget.max_llm_calls:
            return f"LLM call limit reached: {self.usage.llm_calls}/{self.budget.max_llm_calls}"

        if self.usage.tool_calls >= self.budget.max_tool_calls:
            return f"Tool call limit reached: {self.usage.tool_calls}/{self.budget.max_tool_calls}"

        if self.usage.total_tokens >= self.budget.max_total_tokens:
            return f"Token limit reached: {self.usage.total_tokens}/{self.budget.max_total_tokens}"

        if self.usage.total_cost_usd >= self.budget.max_cost_usd:
            return f"Cost limit reached: ${self.usage.total_cost_usd:.3f}/${self.budget.max_cost_usd:.3f}"

        elapsed = self.usage.elapsed_seconds()
        if elapsed >= self.budget.max_wall_time_seconds:
            return f"Time limit reached: {elapsed:.1f}s/{self.budget.max_wall_time_seconds}s"

        return None

    def summary(self) -> dict:
        return {
            "llm_calls": f"{self.usage.llm_calls}/{self.budget.max_llm_calls}",
            "tool_calls": f"{self.usage.tool_calls}/{self.budget.max_tool_calls}",
            "tokens": f"{self.usage.total_tokens}/{self.budget.max_total_tokens}",
            "cost_usd": f"${self.usage.total_cost_usd:.4f}/${self.budget.max_cost_usd:.4f}",
            "elapsed_s": f"{self.usage.elapsed_seconds():.1f}/{self.budget.max_wall_time_seconds}",
        }

"""
智能体评估框架。

除正确率外，还综合衡量效率、工具使用的恰当性
以及推理质量。
"""
from dataclasses import dataclass


@dataclass
class AgentEvalMetrics:
    # 正确率
    final_answer_correct: bool
    partial_credit: float          # 0.0 ~ 1.0（部分得分）

    # 效率
    total_steps: int
    total_tool_calls: int
    total_tokens: int
    total_cost_usd: float
    wall_time_seconds: float

    # 工具使用质量
    unnecessary_tool_calls: int    # 不必要的工具调用次数
    failed_tool_calls: int         # 失败的工具调用次数
    tool_call_accuracy: float      # 以正确输入调用正确工具的比例

    # 推理质量
    reasoning_coherence: float     # 推理的逻辑一致性（0.0 ~ 1.0）
    hallucination_count: int       # 无依据主张的数量

    @property
    def efficiency_score(self) -> float:
        """效率分数：到达正确答案为止使用了多少资源（越少越好）。"""
        if not self.final_answer_correct:
            return 0.0
        # 用量越少越高效 -> 转换为反向分数
        step_penalty = min(self.total_steps / 10, 1.0)
        cost_penalty = min(self.total_cost_usd / 0.10, 1.0)
        return max(0.0, 1.0 - (step_penalty + cost_penalty) / 2)

    @property
    def overall_score(self) -> float:
        """综合分数。"""
        weights = {
            "accuracy": 0.4,
            "efficiency": 0.2,
            "tool_quality": 0.2,
            "reasoning": 0.2,
        }
        accuracy = 1.0 if self.final_answer_correct else self.partial_credit
        return (
            weights["accuracy"] * accuracy
            + weights["efficiency"] * self.efficiency_score
            + weights["tool_quality"] * self.tool_call_accuracy
            + weights["reasoning"] * self.reasoning_coherence
        )