Agent 开发：从 ReAct 循环到生产级智能体系统的专业实践

原创

资源大佬 jzit-top

发布于 2026-09-17 19:21:15

2540

摘要

Agent（智能体）不是“会聊天的机器人”，而是以大模型为推理核心，具备规划、记忆、工具调用与环境交互能力的闭环系统。它的本质是把 LLM 从“文本生成器”升级为“任务执行器”。生产级 Agent 需要解决工具协议、状态管理、错误恢复、成本控制、安全护栏、评估与可观测性等问题。本文从架构、代码、框架、评估、安全与反模式等维度，给出一套可落地的 Agent 开发方法论。

关键词：Agent；ReAct；Function Calling；Plan-and-Execute；Reflexion；多智能体；LangGraph；评估；安全护栏

1. Agent 的定义与核心公式

一个 Agent 可以抽象为：

Agent = LLM + Planning + Memory + Tools + Action Loop

LLM：推理、决策、语言理解与生成；
Planning：任务分解、路径选择、反思修正；
Memory：短期对话、长期知识、任务状态；
Tools：搜索、计算、数据库、API、代码执行；
Action Loop：观察 → 思考 → 行动 → 观察，直到完成或终止。

Agent 与普通 Chain 的区别在于：Chain 是预定义流程，Agent 是动态决定下一步。Agent 的代价是更高的不确定性、延迟、成本和安全风险。

2. 核心架构：感知—规划—行动—记忆

用户输入
  ↓
感知与解析
  ↓
规划器 Planner → 任务列表 / 下一步
  ↓
执行器 Executor → 选择工具 / 生成回答
  ↓
工具层 Tools → 外部世界
  ↓
观察结果 Observation
  ↓
记忆 Memory → 短期 / 长期
  ↓
反思 Critic → 修正计划
  ↓
输出 / 继续循环

关键设计原则：

工具白名单：只暴露必要工具；
状态显式化：任务状态、步骤、预算可追踪；
最大步数：防止无限循环；
可中断：高风险操作需人工确认；
可观测：每步日志、token、成本、延迟；
可评估：任务成功率、步骤效率、工具准确率。

3. 从零实现：工具注册与 Function Calling

生产环境优先使用 Function Calling，而不是让模型输出自由文本再解析。下面实现一个最小可用的工具注册表。

from dataclasses import dataclass
from typing import Callable, Any
import json
import logging

@dataclass
class Tool:
    name: str
    description: str
    func: Callable[..., Any]
    schema: dict

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

    def register(self, tool: Tool):
        self._tools[tool.name] = tool

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

    def to_openai_tools(self) -> list[dict]:
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.schema,
                },
            }
            for t in self._tools.values()
        ]

安全计算器与搜索工具：

import ast
import operator

def safe_calculator(expression: str) -> str:
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.USub: operator.neg,
    }

    def _eval(node):
        if isinstance(node, ast.Expression):
            return _eval(node.body)
        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.BinOp):
            return ops[type(node.op)](_eval(node.left), _eval(node.right))
        if isinstance(node, ast.UnaryOp):
            return ops[type(node.op)](_eval(node.operand))
        raise ValueError("不支持的表达式")

    return str(_eval(ast.parse(expression, mode="eval")))

def search(query: str) -> str:
    # 真实场景接入搜索 API、向量库或内部知识库
    return f"[搜索结果] 关于「{query}」的资料：..."

注册工具：

registry = ToolRegistry()

registry.register(Tool(
    name="search",
    description="搜索互联网或知识库",
    func=search,
    schema={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "搜索关键词"}
        },
        "required": ["query"],
    },
))

registry.register(Tool(
    name="calculator",
    description="执行安全的数学表达式计算",
    func=safe_calculator,
    schema={
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "如 12*(3+4)"}
        },
        "required": ["expression"],
    },
))

4. ReAct Agent：观察—思考—行动循环

ReAct 的核心是让模型在推理与行动之间交替。下面实现一个带 Function Calling 的 Agent。

from openai import OpenAI

class Agent:
    def __init__(
        self,
        client: OpenAI,
        model: str,
        registry: ToolRegistry,
        system_prompt: str,
        max_steps: int = 8,
    ):
        self.client = client
        self.model = model
        self.registry = registry
        self.system_prompt = system_prompt
        self.max_steps = max_steps
        self.messages: list[dict] = [
            {"role": "system", "content": system_prompt}
        ]
        self.logger = logging.getLogger("agent")
        self.step_count = 0

    def run(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})

        for step in range(1, self.max_steps + 1):
            self.step_count = step

            resp = self.client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=self.registry.to_openai_tools(),
                tool_choice="auto",
                temperature=0,
            )
            msg = resp.choices[0].message

            assistant_msg = {"role": "assistant", "content": msg.content or ""}
            if msg.tool_calls:
                assistant_msg["tool_calls"] = [
                    {
                        "id": call.id,
                        "type": "function",
                        "function": {
                            "name": call.function.name,
                            "arguments": call.function.arguments,
                        },
                    }
                    for call in msg.tool_calls
                ]
            self.messages.append(assistant_msg)

            if not msg.tool_calls:
                return msg.content or ""

            for call in msg.tool_calls:
                name = call.function.name
                try:
                    args = json.loads(call.function.arguments or "{}")
                except json.JSONDecodeError:
                    args = {}

                self.logger.info(
                    "step=%s tool=%s args=%s", step, name, args
                )

                try:
                    tool = self.registry.get(name)
                    result = tool.func(**args)
                except Exception as e:
                    result = f"工具执行失败: {e}"

                self.messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": str(result),
                })

        return "达到最大步数，任务未完成。"

使用示例：

client = OpenAI()
agent = Agent(
    client=client,
    model="gpt-4o-mini",
    registry=registry,
    system_prompt=(
        "你是一个严谨的助手。"
        "需要计算时调用 calculator，需要事实时调用 search。"
        "不确定时明确说明，不要编造。"
    ),
)

print(agent.run("计算 12*(3+4)，并搜索 ReAct 论文的核心思想。"))

工程要点：

temperature=0 降低随机性；
每次工具调用都记录日志；
工具异常转为观察结果，而不是中断循环；
最大步数防止无限循环；
工具参数必须校验，不能直接执行危险操作。

5. 规划与反思：Plan-and-Execute 与 Reflexion

ReAct 适合短任务。复杂任务需要显式规划。

5.1 Planner：任务分解

class Planner:
    def __init__(self, client: OpenAI, model: str):
        self.client = client
        self.model = model

    def plan(self, task: str) -> list[str]:
        resp = self.client.chat.completions.create(
            model=self.model,
            temperature=0,
            messages=[
                {
                    "role": "system",
                    "content": (
                        "你是任务规划器。将任务拆成 3-7 个可执行步骤，"
                        "每行一个步骤，不要编号，不要解释。"
                    ),
                },
                {"role": "user", "content": task},
            ],
        )
        text = resp.choices[0].message.content or ""
        return [line.strip() for line in text.splitlines() if line.strip()]

5.2 Executor 与 Critic

class Executor:
    def __init__(self, agent: Agent):
        self.agent = agent

    def execute(self, plan: list[str]) -> str:
        context = ""
        for i, step in enumerate(plan, 1):
            result = self.agent.run(f"当前步骤 {i}: {step}\n已有上下文：{context}")
            context += f"\n步骤{i}结果：{result}"
        return context

class Critic:
    def __init__(self, client: OpenAI, model: str):
        self.client = client
        self.model = model

    def review(self, task: str, result: str) -> str:
        resp = self.client.chat.completions.create(
            model=self.model,
            temperature=0,
            messages=[
                {
                    "role": "system",
                    "content": (
                        "你是评审员。检查结果是否完成任务，"
                        "指出遗漏、错误与改进建议。"
                    ),
                },
                {"role": "user", "content": f"任务：{task}\n结果：{result}"},
            ],
        )
        return resp.choices[0].message.content or ""

5.3 Reflexion：失败后自我修正

def reflexion_loop(agent: Agent, critic: Critic, task: str, max_rounds: int = 3):
    result = agent.run(task)
    for _ in range(max_rounds):
        review = critic.review(task, result)
        if "通过" in review or "完成" in review:
            return result
        result = agent.run(f"根据评审意见修正：\n{review}\n原结果：{result}")
    return result

规划与反思的代价是更多 token 与延迟，适合高价值、可验证任务。

6. 记忆：短期、长期与状态

Agent 记忆分三层：

短期记忆：对话消息列表；
长期记忆：向量库、知识图谱、数据库；
任务状态：计划、步骤、中间结果、预算。

最小长期记忆实现：

import numpy as np

class LongTermMemory:
    def __init__(self, client: OpenAI):
        self.client = client
        self.texts: list[str] = []
        self.vectors: np.ndarray | None = None

    def _embed(self, texts: list[str]) -> np.ndarray:
        resp = self.client.embeddings.create(
            model="text-embedding-3-small", input=texts
        )
        return np.array([d.embedding for d in resp.data], dtype=np.float32)

    def add(self, text: str):
        self.texts.append(text)
        vec = self._embed([text])
        self.vectors = vec if self.vectors is None else np.vstack([self.vectors, vec])

    def search(self, query: str, top_k: int = 3) -> list[str]:
        if self.vectors is None:
            return []
        q = self._embed([query])[0]
        sims = self.vectors @ q / (
            np.linalg.norm(self.vectors, axis=1) * np.linalg.norm(q) + 1e-8
        )
        idx = np.argsort(-sims)[:top_k]
        return [self.texts[i] for i in idx]

在 Agent 每轮开始前检索长期记忆，注入系统提示或用户消息。注意权限隔离与数据脱敏。

7. 多 Agent 协作

多 Agent 不是越多越好。常见角色：

Planner：分解任务；
Researcher：检索与收集；
Executor：调用工具执行；
Critic：评审与修正；
Coordinator：协调与汇总。

简化多 Agent 系统：

class MultiAgentSystem:
    def __init__(self, planner: Planner, executor: Executor, critic: Critic):
        self.planner = planner
        self.executor = executor
        self.critic = critic

    def run(self, task: str) -> str:
        plan = self.planner.plan(task)
        result = self.executor.execute(plan)
        review = self.critic.review(task, result)
        if "通过" in review or "完成" in review:
            return result
        return self.executor.execute([f"根据评审修正：{review}"])

多 Agent 的挑战：

通信开销与上下文膨胀；
角色冲突与责任不清；
成本随 Agent 数量线性增长；
调试困难。

建议：优先单 Agent + 工具 + 规划，确实需要并行或专业分工时再引入多 Agent。

8. 框架与选型

框架	定位	适用场景
LangChain	LLM 应用编排	快速原型、RAG、工具调用
LangGraph	状态图编排	有状态、可中断、多 Agent
AutoGen	多 Agent 对话	研究、协作任务
CrewAI	角色化多 Agent	业务流程模拟
Semantic Kernel	企业级编排	.NET/Python 混合
OpenAI Assistants	托管 Agent	快速接入、托管线程
Dify / Coze	低代码平台	业务人员搭建

选型原则：状态复杂度决定框架。简单工具调用不需要重框架；有状态、可中断、多分支、多 Agent 时，LangGraph 这类状态图更合适。

9. 评估与可观测性

没有评估，就没有 Agent 工程。评估指标：

任务成功率；
平均步数；
工具调用准确率；
工具调用冗余率；
延迟 P50/P95；
token 成本；
安全违规率；
人工接管率。

评估代码示例：

import time

def evaluate_agent(agent: Agent, cases: list[dict]) -> dict:
    results = []
    for case in cases:
        start = time.time()
        try:
            output = agent.run(case["input"])
            success = case["check"](output)
        except Exception as e:
            output = str(e)
            success = False
        results.append({
            "success": success,
            "latency": time.time() - start,
            "steps": agent.step_count,
            "output": output,
        })

    total = len(results)
    return {
        "success_rate": sum(r["success"] for r in results) / total,
        "avg_latency": sum(r["latency"] for r in results) / total,
        "avg_steps": sum(r["steps"] for r in results) / total,
        "details": results,
    }

可观测性：

每步记录 trace_id、step、tool、args、latency、tokens；
接入 OpenTelemetry、LangSmith、LangFuse 或自建看板；
对失败步骤保留完整上下文，便于回放。

10. 安全与生产化

Agent 的安全风险高于普通 LLM 应用：

提示注入：用户诱导模型调用危险工具；
工具越权：访问未授权数据或系统；
数据泄露：上下文包含敏感信息；
无限循环：消耗 token 与时间；
幻觉行动：基于错误事实执行操作。

安全护栏：

DANGEROUS_TOOLS = {"shell", "delete_file", "execute_sql"}

def guard_tool_call(tool_name: str, args: dict, user_role: str):
    if tool_name in DANGEROUS_TOOLS and user_role != "admin":
        raise PermissionError(f"无权调用工具: {tool_name}")
    if tool_name == "calculator":
        expr = args.get("expression", "")
        if any(x in expr for x in ["__", "import", "open", "eval"]):
            raise ValueError("非法表达式")
    return True

生产化清单：

工具白名单与权限模型；
参数校验与输入过滤；
最大步数、最大 token、最大成本；
超时、重试、熔断、降级；
人工确认高风险操作；
全链路日志与审计；
评估集与回归测试；
灰度发布与回滚；
数据脱敏与隔离；
安全事件响应预案。

11. 常见反模式

把 Agent 当万能，简单任务也上多 Agent；
没有最大步数与预算控制；
让模型直接执行 Shell、SQL、文件删除；
工具描述模糊，导致误调用；
不做评估就上线；
没有日志与回放能力；
上下文无限增长；
用 Agent 解决知识更新问题，而不是 RAG；
多 Agent 角色重叠，成本失控；
忽视提示注入与权限隔离。

12. 结论

Agent 开发是 LLM 工程的高级形态。它把模型从“生成文本”推进到“执行任务”，但代价是不确定性、成本、延迟与安全风险。专业 Agent 系统必须建立在清晰的工具协议、显式状态、规划与反思、分层记忆、评估与可观测性、安全护栏之上。代码上，从 Function Calling 与 ReAct 循环起步，逐步引入 Planner、Critic、长期记忆和多 Agent；工程上，先保证可控、可测、可回滚，再追求自治与智能。真正生产级的 Agent，不是最像人的 Agent，而是最可靠、最可解释、最可治理的 Agent。

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

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

agent