Agent Loop 深度解析：从 ReAct 到生产级循环的工程内核

原创

资源大佬 jzit-top

发布于 2026-09-19 17:50:08

1510

摘要

Agent Loop 是 AI Agent 的心脏。它不是简单的 while True 调 LLM，而是由推理、行动、观察、状态更新、终止判定五个环节组成的闭环控制系统。产品级 Agent 与 Demo 的差距，几乎全部体现在循环工程上：如何防止无限循环、如何压缩上下文、如何处理工具失败、如何并发执行、如何可观测。本文从控制论视角拆解 Agent Loop，对比 ReAct、Plan-and-Execute、Reflexion 三种范式，并用 Python 实现一个可运行、可扩展、带生产级特性的 Agent Loop 内核。

一、Agent Loop 的本质：一个闭环控制系统

如果把 Agent 类比为控制系统，它的结构如下：

      ┌──────────────────────────────────────┐
      │                                      │
      ▼                                      │
  ┌────────┐   ┌────────┐   ┌────────┐   ┌────────┐
  │ 推理   │──▶│ 行动   │──▶│ 观察   │──▶│ 状态   │
  │ Reason │   │  Act   │   │Observe │   │Update  │
  └────────┘   └────────┘   └────────┘   └────────┘
      ▲                                      │
      │           ┌────────────┐             │
      └───────────│ 终止判定    │◀────────────┘
                  │Termination │
                  └────────────┘

与传统程序的区别：

维度	传统程序	Agent Loop
控制流	静态、确定	动态、由 LLM 决定
循环次数	编译期确定	运行期决定
错误处理	try/catch	观察错误 → 重新推理
状态	显式变量	消息历史 + 外部状态
终止	return	无工具调用 / 达上限 / 显式结束

核心洞察：Agent Loop 的难点不在"循环"，而在"何时停、如何记、错了怎么办"。

二、三种主流 Loop 范式

2.1 ReAct：推理与行动交替

ReAct（Reason + Act）是最经典的范式：

Thought: 我需要查天气
Action: get_weather(city="北京")
Observation: 晴，18°C
Thought: 已有信息，可以回答
Answer: 北京今天晴，18°C

特点：

每一步都"想一下再做"；
适合工具调用链短的场景；
缺点是长任务容易跑偏。

2.2 Plan-and-Execute：先规划再执行

Plan: [查天气, 查温度趋势, 生成建议]
Execute step 1 → ...
Execute step 2 → ...
Execute step 3 → ...

特点：

先分解任务，再逐步执行；
适合多步骤、可预见的任务；
缺点是计划可能过时，需重新规划。

2.3 Reflexion：执行后反思

Attempt 1 → 失败 → Reflection: 原因分析
Attempt 2 → 改进 → 成功

特点：

引入自我批评；
适合代码生成、复杂推理；
成本高，需控制反思轮次。

生产建议：多数场景用 ReAct 为骨架，复杂任务叠加 Plan，关键任务叠加 Reflexion。

三、Agent Loop 的五个核心环节

3.1 推理（Reason）

输入：消息历史 + 工具定义输出：自然语言思考 + 工具调用意图

关键工程点：

系统提示词要明确"何时用工具、何时直接答"；
工具描述要精确，LLM 靠描述选择工具；
temperature 要低（0.1 左右），减少随机性。

3.2 行动（Act）

解析 LLM 输出的工具调用，执行。

关键工程点：

参数校验（schema、必填、类型）；
超时控制；
并发执行独立工具；
结果标准化。

3.3 观察（Observe）

把工具结果回填到消息历史。

关键工程点：

长结果要截断或摘要；
错误要结构化（{"error": "..."}），便于 LLM 理解；
敏感信息要脱敏。

3.4 状态更新（Update）

维护消息历史、外部状态、轮次计数。

关键工程点：

上下文窗口管理（滑动窗口、摘要）；
状态持久化（跨请求）；
记忆分层（短期/长期）。

3.5 终止判定（Termination）

决定循环是否继续。

五类终止条件：

LLM 无工具调用（正常结束）；
达到最大轮次（保护）；
达到 token 预算（成本保护）；
达到超时（SLA 保护）；
显式终止工具（如 finish）。

四、实战：生产级 Agent Loop 内核

4.1 项目结构

agent-loop/
├── agent_loop/
│   ├── __init__.py
│   ├── types.py          # 数据结构
│   ├── context.py        # 上下文管理
│   ├── tools.py          # 工具注册
│   ├── executor.py       # 工具执行器
│   ├── llm.py            # LLM 客户端
│   ├── loop.py           # Agent Loop 内核
│   └── observability.py  # 可观测性
├── requirements.txt
└── demo.py

4.2 requirements.txt

httpx>=0.27.0
pydantic>=2.5.0
tiktoken>=0.7.0

4.3 数据结构：types.py

"""Agent Loop 核心数据结构。"""

from dataclasses import dataclass, field
from enum import Enum
from typing import Any


class Role(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    TOOL = "tool"


class TerminationReason(str, Enum):
    NO_TOOL_CALL = "no_tool_call"       # LLM 主动结束
    MAX_ROUNDS = "max_rounds"           # 达到最大轮次
    TOKEN_BUDGET = "token_budget"       # 达到 token 预算
    TIMEOUT = "timeout"                 # 超时
    EXPLICIT_FINISH = "explicit_finish" # 显式 finish 工具
    ERROR = "error"                     # 不可恢复错误


@dataclass
class Message:
    role: Role
    content: str = ""
    name: str | None = None
    tool_call_id: str | None = None
    tool_calls: list[dict] | None = None

    def to_dict(self) -> dict:
        d: dict[str, Any] = {"role": self.role.value}
        if self.content:
            d["content"] = self.content
        if self.name:
            d["name"] = self.name
        if self.tool_call_id:
            d["tool_call_id"] = self.tool_call_id
        if self.tool_calls:
            d["tool_calls"] = self.tool_calls
        return d


@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict[str, Any]


@dataclass
class ToolResult:
    tool_call_id: str
    name: str
    ok: bool
    content: str
    duration_ms: float = 0.0
    error: str | None = None


@dataclass
class LoopResult:
    answer: str
    reason: TerminationReason
    rounds: int
    total_tokens: int
    duration_ms: float
    tool_calls: list[ToolCall] = field(default_factory=list)
    tool_results: list[ToolResult] = field(default_factory=list)
    trace: list[dict] = field(default_factory=list)

4.4 可观测性：observability.py

"""轻量级 Trace，记录循环每一轮。"""

import logging
import time
from dataclasses import dataclass, field, asdict

logger = logging.getLogger("agent_loop")
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)


@dataclass
class RoundTrace:
    round_index: int
    reason_text: str = ""
    tool_calls: list[dict] = field(default_factory=list)
    tool_results: list[dict] = field(default_factory=list)
    tokens: int = 0
    duration_ms: float = 0.0


class LoopTracer:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.rounds: list[RoundTrace] = []
        self.start_ts = time.time()

    def new_round(self, index: int) -> RoundTrace:
        rt = RoundTrace(round_index=index)
        self.rounds.append(rt)
        return rt

    def finish(self, reason: str):
        total_ms = (time.time() - self.start_ts) * 1000
        logger.info(
            "session=%s reason=%s rounds=%d total_ms=%.1f",
            self.session_id, reason, len(self.rounds), total_ms,
        )

    def dump(self) -> list[dict]:
        return [asdict(r) for r in self.rounds]

4.5 上下文管理：context.py

"""上下文管理：滑动窗口 + 摘要压缩 + token 预算。"""

from typing import Callable

from .types import Message, Role

try:
    import tiktoken
    _ENC = tiktoken.get_encoding("cl100k_base")

    def count_tokens(text: str) -> int:
        return len(_ENC.encode(text))
except Exception:
    def count_tokens(text: str) -> int:
        # 粗略估算：中文约 1.5 字符/token，英文约 4 字符/token
        return max(1, len(text) // 2)


class ContextManager:
    """
    上下文窗口管理策略：
    1. 永远保留 system 消息
    2. 永远保留最近 N 条消息
    3. 中间部分按需摘要
    4. 超过 token 预算时触发压缩
    """

    def __init__(
        self,
        max_tokens: int = 8192,
        keep_recent: int = 8,
        summarize_fn: Callable[[list[Message]], str] | None = None,
    ):
        self.max_tokens = max_tokens
        self.keep_recent = keep_recent
        self.summarize_fn = summarize_fn
        self._summary: str | None = None

    def total_tokens(self, messages: list[Message]) -> int:
        total = 0
        for m in messages:
            total += count_tokens(m.content or "")
            if m.tool_calls:
                for tc in m.tool_calls:
                    fn = tc.get("function", {})
                    total += count_tokens(
                        fn.get("name", "") + fn.get("arguments", "")
                    )
        return total

    def compact(self, messages: list[Message]) -> list[Message]:
        """超过预算时压缩上下文，返回新列表。"""
        if self.total_tokens(messages) <= self.max_tokens:
            return messages

        system_msgs = [m for m in messages if m.role == Role.SYSTEM]
        rest = [m for m in messages if m.role != Role.SYSTEM]

        if len(rest) <= self.keep_recent:
            return messages

        old = rest[: -self.keep_recent]
        recent = rest[-self.keep_recent:]

        if self.summarize_fn:
            self._summary = self.summarize_fn(old)
        else:
            self._summary = self._fallback_summary(old)

        summary_msg = Message(
            role=Role.SYSTEM,
            content=f"[历史摘要] {self._summary}",
        )
        return system_msgs + [summary_msg] + recent

    @staticmethod
    def _fallback_summary(messages: list[Message]) -> str:
        """无 LLM 时的降级摘要：保留关键角色和工具名。"""
        parts = []
        for m in messages:
            if m.role == Role.USER and m.content:
                parts.append(f"用户: {m.content[:80]}")
            elif m.role == Role.ASSISTANT and m.tool_calls:
                names = [tc["function"]["name"] for tc in m.tool_calls]
                parts.append(f"调用: {','.join(names)}")
        return " | ".join(parts[-6:]) if parts else "（无重要历史）"

4.6 工具注册：tools.py

"""工具注册表：注册、校验、schema 导出。"""

import inspect
from typing import Any, Callable

from .types import ToolCall


class ToolError(Exception):
    pass


class Tool:
    def __init__(
        self,
        name: str,
        description: str,
        parameters: dict[str, Any],
        func: Callable,
        dangerous: bool = False,
    ):
        self.name = name
        self.description = description
        self.parameters = parameters
        self.func = func
        self.dangerous = dangerous

    def validate(self, args: dict[str, Any]) -> None:
        props = self.parameters.get("properties", {})
        required = set(self.parameters.get("required", []))

        missing = required - set(args)
        if missing:
            raise ToolError(f"缺少必填参数: {missing}")

        unknown = set(args) - set(props)
        if unknown:
            raise ToolError(f"未知参数: {unknown}")

        for key, value in args.items():
            expected = props[key].get("type")
            if expected == "string" and not isinstance(value, str):
                raise ToolError(f"参数 {key} 应为 string")
            if expected == "integer" and not isinstance(value, int):
                raise ToolError(f"参数 {key} 应为 integer")
            if expected == "number" and not isinstance(value, (int, float)):
                raise ToolError(f"参数 {key} 应为 number")
            if expected == "boolean" and not isinstance(value, bool):
                raise ToolError(f"参数 {key} 应为 boolean")
            if "enum" in props[key] and value not in props[key]["enum"]:
                raise ToolError(
                    f"参数 {key} 必须是 {props[key]['enum']} 之一"
                )

    def to_schema(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }

    async def run(self, args: dict[str, Any]) -> Any:
        self.validate(args)
        result = self.func(**args)
        if inspect.isawaitable(result):
            result = await result
        return result


class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"工具 {tool.name} 已注册")
        self._tools[tool.name] = tool

    def get(self, name: str) -> Tool:
        if name not in self._tools:
            raise ToolError(f"未知工具: {name}")
        return self._tools[name]

    def schemas(self) -> list[dict]:
        return [t.to_schema() for t in self._tools.values()]

    def names(self) -> list[str]:
        return list(self._tools.keys())


def parse_tool_calls(raw: list[dict]) -> list[ToolCall]:
    """把 LLM 返回的 tool_calls 标准化。"""
    import json
    import uuid

    out: list[ToolCall] = []
    for tc in raw:
        fn = tc.get("function", {})
        args_raw = fn.get("arguments", "{}")
        if isinstance(args_raw, str):
            try:
                args = json.loads(args_raw) if args_raw else {}
            except json.JSONDecodeError:
                args = {}
        else:
            args = args_raw or {}
        out.append(ToolCall(
            id=tc.get("id") or f"call_{uuid.uuid4().hex[:8]}",
            name=fn.get("name", ""),
            arguments=args,
        ))
    return out

4.7 工具执行器：executor.py

"""工具执行器：并发执行、超时、错误包装。"""

import asyncio
import time

from .tools import ToolRegistry, ToolError
from .types import ToolCall, ToolResult


class ToolExecutor:
    def __init__(self, registry: ToolRegistry,
                 default_timeout: float = 10.0,
                 max_concurrency: int = 4):
        self.registry = registry
        self.default_timeout = default_timeout
        self.semaphore = asyncio.Semaphore(max_concurrency)

    async def execute(self, call: ToolCall) -> ToolResult:
        start = time.time()
        try:
            async with self.semaphore:
                tool = self.registry.get(call.name)
                result = await asyncio.wait_for(
                    tool.run(call.arguments),
                    timeout=self.default_timeout,
                )
            return ToolResult(
                tool_call_id=call.id,
                name=call.name,
                ok=True,
                content=str(result),
                duration_ms=(time.time() - start) * 1000,
            )
        except asyncio.TimeoutError:
            return ToolResult(
                tool_call_id=call.id,
                name=call.name,
                ok=False,
                content=f"[TIMEOUT] 工具 {call.name} 超过 "
                        f"{self.default_timeout}s",
                duration_ms=(time.time() - start) * 1000,
                error="timeout",
            )
        except ToolError as e:
            return ToolResult(
                tool_call_id=call.id,
                name=call.name,
                ok=False,
                content=f"[ERROR] {e}",
                duration_ms=(time.time() - start) * 1000,
                error=str(e),
            )
        except Exception as e:
            return ToolResult(
                tool_call_id=call.id,
                name=call.name,
                ok=False,
                content=f"[ERROR] {type(e).__name__}: {e}",
                duration_ms=(time.time() - start) * 1000,
                error=str(e),
            )

    async def execute_batch(self, calls: list[ToolCall]) -> list[ToolResult]:
        """并发执行多个工具调用。"""
        if not calls:
            return []
        return await asyncio.gather(*[self.execute(c) for c in calls])

4.8 LLM 客户端：llm.py

"""LLM 客户端：OpenAI 兼容 + Mock。"""

import json
import os

import httpx

from .types import Message


class LLMClient:
    def __init__(
        self,
        mode: str = "mock",
        base_url: str = "https://api.openai.com/v1",
        api_key: str | None = None,
        model: str = "gpt-4o-mini",
    ):
        self.mode = mode
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key or os.getenv("OPENAI_API_KEY", "")
        self.model = model

    async def chat(
        self,
        messages: list[Message],
        tools: list[dict] | None = None,
    ) -> tuple[str, list[dict], int]:
        """返回 (content, tool_calls, tokens)。"""
        if self.mode == "mock":
            return self._mock(messages, tools)
        return await self._openai(messages, tools)

    async def _openai(
        self,
        messages: list[Message],
        tools: list[dict] | None,
    ) -> tuple[str, list[dict], int]:
        payload: dict = {
            "model": self.model,
            "messages": [m.to_dict() for m in messages],
            "temperature": 0.1,
        }
        if tools:
            payload["tools"] = tools
            payload["tool_choice"] = "auto"

        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        async with httpx.AsyncClient(timeout=60) as client:
            resp = await client.post(
                f"{self.base_url}/chat/completions",
                json=payload, headers=headers,
            )
            resp.raise_for_status()
            data = resp.json()

        msg = data["choices"][0]["message"]
        tokens = data.get("usage", {}).get("total_tokens", 0)
        return (
            msg.get("content") or "",
            msg.get("tool_calls") or [],
            tokens,
        )

    def _mock(
        self,
        messages: list[Message],
        tools: list[dict] | None,
    ) -> tuple[str, list[dict], int]:
        """Mock：根据最近一条用户消息，走固定脚本。"""
        user_msg = ""
        for m in reversed(messages):
            if m.role.value == "user":
                user_msg = m.content or ""
                break

        # 已有工具结果 → 生成最终回答
        has_tool_result = any(
            m.role.value == "tool" for m in messages[-3:]
        )
        if has_tool_result:
            last_tool = next(
                (m for m in reversed(messages) if m.role.value == "tool"),
                None,
            )
            content = (
                f"根据工具结果：{last_tool.content if last_tool else ''}"
                f"，我已经完成了你的请求。"
            )
            return content, [], 100

        # 首次调用 → 触发工具
        if "天气" in user_msg:
            return "", [{
                "id": "call_mock_1",
                "function": {
                    "name": "get_weather",
                    "arguments": json.dumps({"city": "北京"}),
                },
            }], 80

        if "计算" in user_msg or "算" in user_msg:
            return "", [{
                "id": "call_mock_2",
                "function": {
                    "name": "calculator",
                    "arguments": json.dumps({"expression": "23*17"}),
                },
            }], 80

        if "搜索" in user_msg:
            return "", [{
                "id": "call_mock_3",
                "function": {
                    "name": "search",
                    "arguments": json.dumps({"query": user_msg}),
                },
            }], 80

        return "我是 Agent Loop 演示，试试问天气、计算或搜索。", [], 50

4.9 Agent Loop 内核：loop.py

"""Agent Loop 内核：ReAct 循环 + 多重终止保护。"""

import time

from .context import ContextManager
from .executor import ToolExecutor
from .llm import LLMClient
from .observability import LoopTracer
from .tools import ToolRegistry, parse_tool_calls
from .types import (
    LoopResult, Message, Role, TerminationReason,
)

DEFAULT_SYSTEM_PROMPT = """你是一个能调用工具的 AI 助手。

规则：
1. 需要外部信息或执行操作时，调用工具，不要编造。
2. 每次只调用必要的工具。
3. 拿到工具结果后，用简洁中文回答用户。
4. 如果不需要工具，直接回答。
5. 如果任务已完成，不要再调用工具。
"""


class AgentLoop:
    def __init__(
        self,
        llm: LLMClient,
        registry: ToolRegistry,
        system_prompt: str = DEFAULT_SYSTEM_PROMPT,
        max_rounds: int = 6,
        max_tokens: int = 8192,
        timeout_seconds: float = 120.0,
        tool_timeout: float = 10.0,
        summarize_fn=None,
    ):
        self.llm = llm
        self.registry = registry
        self.system_prompt = system_prompt
        self.max_rounds = max_rounds
        self.timeout_seconds = timeout_seconds
        self.executor = ToolExecutor(registry, default_timeout=tool_timeout)
        self.context = ContextManager(
            max_tokens=max_tokens,
            summarize_fn=summarize_fn,
        )
        self.history: list[Message] = [
            Message(role=Role.SYSTEM, content=system_prompt)
        ]

    def reset(self) -> None:
        self.history = [
            Message(role=Role.SYSTEM, content=self.system_prompt)
        ]

    async def run(
        self,
        user_input: str,
        session_id: str = "default",
    ) -> LoopResult:
        tracer = LoopTracer(session_id)
        start = time.time()

        self.history.append(Message(role=Role.USER, content=user_input))

        all_calls = []
        all_results = []
        total_tokens = 0
        answer = ""
        reason = TerminationReason.MAX_ROUNDS

        for round_idx in range(self.max_rounds):
            # 超时保护
            if time.time() - start > self.timeout_seconds:
                reason = TerminationReason.TIMEOUT
                answer = "（请求超时，请简化问题或稍后重试）"
                break

            # 上下文压缩
            self.history = self.context.compact(self.history)

            rt = tracer.new_round(round_idx)
            round_start = time.time()

            # 1. 推理
            try:
                content, raw_calls, tokens = await self.llm.chat(
                    messages=self.history,
                    tools=self.registry.schemas(),
                )
            except Exception as e:
                reason = TerminationReason.ERROR
                answer = f"（LLM 调用失败: {type(e).__name__}: {e}）"
                break

            total_tokens += tokens
            rt.tokens = tokens
            rt.reason_text = content

            # 2. 终止判定：无工具调用
            if not raw_calls:
                answer = content or "（无回复）"
                reason = TerminationReason.NO_TOOL_CALL
                self.history.append(
                    Message(role=Role.ASSISTANT, content=answer)
                )
                rt.duration_ms = (time.time() - round_start) * 1000
                break

            # 3. 解析工具调用
            calls = parse_tool_calls(raw_calls)
            all_calls.extend(calls)
            rt.tool_calls = [
                {"name": c.name, "arguments": c.arguments} for c in calls
            ]

            # 4. 记录 assistant 意图
            self.history.append(Message(
                role=Role.ASSISTANT,
                content=content or "",
                tool_calls=raw_calls,
            ))

            # 5. 行动 + 观察（并发执行）
            results = await self.executor.execute_batch(calls)
            all_results.extend(results)

            for res in results:
                # 显式 finish 工具
                if res.name == "finish" and res.ok:
                    reason = TerminationReason.EXPLICIT_FINISH
                    answer = res.content
                    rt.duration_ms = (time.time() - round_start) * 1000
                    return LoopResult(
                        answer=answer,
                        reason=reason,
                        rounds=round_idx + 1,
                        total_tokens=total_tokens,
                        duration_ms=(time.time() - start) * 1000,
                        tool_calls=all_calls,
                        tool_results=all_results,
                        trace=tracer.dump(),
                    )

                # 回填 tool 消息
                self.history.append(Message(
                    role=Role.TOOL,
                    name=res.name,
                    tool_call_id=res.tool_call_id,
                    content=res.content,
                ))

            rt.tool_results = [
                {"name": r.name, "ok": r.ok,
                 "duration_ms": round(r.duration_ms, 2)}
                for r in results
            ]
            rt.duration_ms = (time.time() - round_start) * 1000

            # token 预算保护
            if total_tokens >= self.context.max_tokens:
                reason = TerminationReason.TOKEN_BUDGET
                answer = "（达到 token 预算，已停止）"
                break

        tracer.finish(reason.value)

        return LoopResult(
            answer=answer,
            reason=reason,
            rounds=len(tracer.rounds),
            total_tokens=total_tokens,
            duration_ms=(time.time() - start) * 1000,
            tool_calls=all_calls,
            tool_results=all_results,
            trace=tracer.dump(),
        )

4.10 入口：init.py

from .loop import AgentLoop
from .llm import LLMClient
from .tools import Tool, ToolRegistry
from .types import (
    LoopResult, Message, Role,
    ToolCall, ToolResult, TerminationReason,
)

__all__ = [
    "AgentLoop", "LLMClient", "Tool", "ToolRegistry",
    "LoopResult", "Message", "Role",
    "ToolCall", "ToolResult", "TerminationReason",
]
__version__ = "0.1.0"

4.11 演示：demo.py

"""Agent Loop 演示：注册工具并运行。"""

import asyncio
import os

from agent_loop import AgentLoop, LLMClient, Tool, ToolRegistry


def build_registry() -> ToolRegistry:
    reg = ToolRegistry()

    def get_weather(city: str) -> str:
        data = {"北京": "晴，18°C", "上海": "多云，22°C"}
        return data.get(city, f"{city}: 暂无数据")

    def calculator(expression: str) -> str:
        allowed = set("0123456789+-*/(). ")
        if not set(expression) <= allowed:
            raise ValueError("表达式含非法字符")
        return str(eval(expression, {"__builtins__": {}}, {}))

    async def search(query: str) -> str:
        await asyncio.sleep(0.05)
        return f"[搜索结果] 关于“{query}”的前 3 条摘要..."

    def finish(answer: str) -> str:
        return answer

    reg.register(Tool(
        name="get_weather",
        description="查询城市天气",
        parameters={
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名"},
            },
            "required": ["city"],
        },
        func=get_weather,
    ))

    reg.register(Tool(
        name="calculator",
        description="计算数学表达式",
        parameters={
            "type": "object",
            "properties": {
                "expression": {"type": "string"},
            },
            "required": ["expression"],
        },
        func=calculator,
    ))

    reg.register(Tool(
        name="search",
        description="搜索信息",
        parameters={
            "type": "object",
            "properties": {
                "query": {"type": "string"},
            },
            "required": ["query"],
        },
        func=search,
    ))

    reg.register(Tool(
        name="finish",
        description="任务完成时调用，传入最终答案",
        parameters={
            "type": "object",
            "properties": {
                "answer": {"type": "string"},
            },
            "required": ["answer"],
        },
        func=finish,
    ))

    return reg


async def main():
    mode = os.getenv("AGENT_MODE", "")
    if not mode:
        mode = "openai" if os.getenv("OPENAI_API_KEY") else "mock"

    llm = LLMClient(
        mode=mode,
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
        api_key=os.getenv("OPENAI_API_KEY"),
        model=os.getenv("AGENT_MODEL", "gpt-4o-mini"),
    )

    loop = AgentLoop(
        llm=llm,
        registry=build_registry(),
        max_rounds=6,
    )

    print(f"=== Agent Loop 演示（模式: {mode}）===")
    print("输入问题，空行退出。\n")

    while True:
        try:
            user_input = input("你 > ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if not user_input:
            break

        result = await loop.run(user_input, session_id="cli")
        print(f"\nAgent > {result.answer}")
        print(
            f"[终止原因: {result.reason.value} | "
            f"轮次: {result.rounds} | "
            f"tokens: {result.total_tokens} | "
            f"耗时: {result.duration_ms:.0f}ms]\n"
        )

        # 打印 trace
        for r in result.trace:
            tools = ",".join(t["name"] for t in r.tool_calls) or "-"
            print(
                f"  · 轮次 {r.round_index}: "
                f"tools=[{tools}] "
                f"tokens={r.tokens} "
                f"耗时={r.duration_ms:.0f}ms"
            )
        print()


if __name__ == "__main__":
    asyncio.run(main())

4.12 运行

# Mock 模式（无需 API Key）
python demo.py

# 真实 LLM
export OPENAI_API_KEY=sk-xxxx
export AGENT_MODEL=gpt-4o-mini
python demo.py

预期输出：

=== Agent Loop 演示（模式: mock）===
输入问题，空行退出。

你 > 北京天气怎么样

Agent > 根据工具结果：晴，18°C，我已经完成了你的请求。
[终止原因: no_tool_call | 轮次: 2 | tokens: 180 | 耗时: 12ms]

  · 轮次 0: tools=[get_weather] tokens=80 耗时=5ms
  · 轮次 1: tools=[-] tokens=100 耗时=7ms

五、生产级 Loop 的关键工程问题

5.1 防止无限循环

五重保护（已在代码中实现）：

1. max_rounds          # 轮次上限
2. timeout_seconds     # 总超时
3. max_tokens          # token 预算
4. no_tool_call        # 正常结束
5. explicit_finish     # 显式结束工具

5.2 上下文压缩

三种策略叠加：

策略	触发条件	效果
滑动窗口	消息数超限	保留最近 N 条
摘要压缩	token 超预算	历史变一段摘要
工具结果截断	单条结果过长	截断到 N 字符

代码中 ContextManager.compact 已实现前两种。

5.3 工具失败恢复

关键设计：错误不抛出循环，而是作为观察回填。

# 错误结果示例
{"error": "timeout", "content": "[TIMEOUT] 工具 get_weather 超过 10s"}

LLM 看到这个结果后，可以：

重试同一工具；
换一个工具；
告知用户失败。

5.4 并发执行

独立工具调用应并发执行。代码中 ToolExecutor.execute_batch 用 asyncio.gather 实现，配合 Semaphore 限流。

注意：有依赖关系的工具不能并发（如"先查订单再退款"），需 LLM 分轮调用。

5.5 状态持久化

生产环境需把 self.history 存到 Redis 或数据库，支持：

跨请求延续会话；
多实例部署；
会话回放。

5.6 可观测性

每轮记录：

推理文本；
工具调用与结果；
token 消耗；
耗时。

代码中 LoopTracer 已实现，生产环境应落地到 ClickHouse 或 ES。

六、进阶：Plan-and-Execute 变体

在 ReAct 基础上增加规划阶段：

class PlanAndExecuteLoop(AgentLoop):
    """先规划，再执行。适合多步骤任务。"""

    async def run(self, user_input: str, session_id: str = "default"):
        # 阶段 1：规划
        plan_prompt = (
            f"用户请求：{user_input}\n\n"
            f"可用工具：{self.registry.names()}\n\n"
            f"请分解为 3-7 个步骤，每步一行，格式：\n"
            f"1. [步骤描述]"
        )
        content, _, tokens = await self.llm.chat(
            messages=[Message(role=Role.USER, content=plan_prompt)],
            tools=None,
        )
        steps = [
            line.strip() for line in content.split("\n")
            if line.strip() and line.strip()[0].isdigit()
        ]

        # 阶段 2：逐步执行
        results = []
        for i, step in enumerate(steps):
            step_input = f"当前步骤 {i+1}/{len(steps)}：{step}"
            result = await super().run(step_input, session_id)
            results.append({
                "step": step,
                "answer": result.answer,
                "reason": result.reason.value,
            })
            if result.reason == TerminationReason.ERROR:
                break

        # 阶段 3：汇总
        summary = "\n".join(
            f"{i+1}. {r['step']} → {r['answer'][:100]}"
            for i, r in enumerate(results)
        )
        return LoopResult(
            answer=summary,
            reason=TerminationReason.NO_TOOL_CALL,
            rounds=len(results),
            total_tokens=tokens,
            duration_ms=0,
        )

七、测试策略

# test_loop.py
import asyncio
import pytest

from agent_loop import AgentLoop, LLMClient, Tool, ToolRegistry
from agent_loop.types import TerminationReason


def build_registry():
    reg = ToolRegistry()
    reg.register(Tool(
        name="echo",
        description="回显输入",
        parameters={
            "type": "object",
            "properties": {"text": {"type": "string"}},
            "required": ["text"],
        },
        func=lambda text: text,
    ))
    return reg


@pytest.mark.asyncio
async def test_mock_loop_terminates():
    loop = AgentLoop(
        llm=LLMClient(mode="mock"),
        registry=build_registry(),
        max_rounds=3,
    )
    result = await loop.run("你好")
    assert result.reason in (
        TerminationReason.NO_TOOL_CALL,
        TerminationReason.MAX_ROUNDS,
    )
    assert result.rounds <= 3


@pytest.mark.asyncio
async def test_tool_timeout():
    async def slow_tool(text: str) -> str:
        await asyncio.sleep(10)
        return text

    reg = ToolRegistry()
    reg.register(Tool(
        name="slow",
        description="慢工具",
        parameters={
            "type": "object",
            "properties": {"text": {"type": "string"}},
            "required": ["text"],
        },
        func=slow_tool,
    ))
    loop = AgentLoop(
        llm=LLMClient(mode="mock"),
        registry=reg,
        tool_timeout=0.1,
    )
    # 直接测试 executor
    from agent_loop.types import ToolCall
    result = await loop.executor.execute(
        ToolCall(id="1", name="slow", arguments={"text": "hi"})
    )
    assert not result.ok
    assert result.error == "timeout"

运行：

pytest test_loop.py -v

八、性能与成本优化

优化方向	手段	收益
减少轮次	提示词明确工具选择	降 token、降延迟
并发工具	asyncio.gather	降端到端延迟
上下文压缩	滑动窗口 + 摘要	显著降本
结果截断	工具结果限长	防上下文爆炸
模型路由	简单任务小模型	平衡质量成本
缓存	相同调用结果缓存	减少重复

九、结语

Agent Loop 是 AI Agent 的心脏，但它的复杂度远高于"循环调用 LLM"。生产级 Loop 必须解决五个问题：

何时停：多重终止条件，防止无限循环；
如何记：上下文压缩，控制 token 成本；
错了怎么办：错误作为观察回填，让 LLM 决策；
如何快：并发执行独立工具；
如何可观测：每轮 trace，可回放可调试。

本文给出的实现是一个可运行、可扩展、带生产级保护的 Agent Loop 内核。它不是最简的 Demo，但是能上生产的最小闭环。真正落地时，你还需要把上下文存到 Redis、把 trace 落到数据库、把工具接上真实系统，并在每一步加好监控与限流。

Loop 的本质，是把不确定性关进确定性的笼子里。 LLM 负责推理，工程负责边界。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

agent