AI Agent 使用避坑指南｜附源码

随着大模型能力的快速提升，AI Agent（智能体）已经从“概念演示”逐渐进入真实业务场景：自动客服、数据分析助手、代码生成、运营自动化、知识库问答、流程编排、个人效率工具……很多团队都开始尝试把大模型接入自己的系统，让它不仅能“回答问题”，还能“调用工具、执行任务、完成流程”。

但真正落地之后，很多人会发现：AI Agent 看起来很智能，用起来却很容易翻车。它可能调用错工具、编造不存在的信息、执行危险操作、陷入循环、输出不稳定、成本失控，甚至在生产环境中造成业务风险。

本文将从实战角度，总结 AI Agent 使用中的常见坑点，并给出一套可运行的简化版 Agent 源码示例，帮助你理解一个 Agent 的基本结构、工具调用方式、风险控制思路和工程化落地要点。

一、什么是 AI Agent？

简单来说，AI Agent 是一种基于大模型的“任务执行系统”。它不仅能理解用户输入，还可以根据目标进行规划、调用工具、获取外部信息，并持续迭代直到完成任务。

一个典型 AI Agent 通常包含以下几个核心模块：

1. 大语言模型（LLM）
   用于理解用户意图、推理、规划、生成结果。

2. 工具系统（Tools）
   例如搜索工具、数据库查询工具、代码执行工具、邮件发送工具、订单查询工具等。

3. 记忆系统（Memory）
   保存历史对话、用户偏好、长期知识或任务上下文。

4. 规划与执行机制（Planning & Acting）
   将复杂任务拆解为多个步骤，并逐步完成。

5. 观察与反馈（Observation）
   工具调用后得到结果，再交给模型继续判断下一步。

6. 安全与约束（Guardrails）
   限制 Agent 的行为边界，避免危险操作和错误输出。

可以把 AI Agent 理解为：

大模型 + 工具调用 + 上下文记忆 + 任务规划 + 执行反馈 + 安全控制。

二、为什么 AI Agent 容易踩坑？

很多团队在刚开始做 Agent 时，容易高估大模型的“自主能力”，低估工程系统的重要性。

大模型本质上是一个概率生成模型，它并不真正理解系统边界，也不会天然知道哪些操作危险、哪些数据可信、哪些工具可以调用。因此，如果你只是简单地把工具列表塞给模型，然后让它自由发挥，就很容易出现问题。

例如：

• 用户说“帮我删除所有测试数据”，Agent 真的调用了删除接口；
• 工具返回异常，Agent 不处理，继续编造结果；
• 数据库查不到信息，Agent 却假装查到了；
• 任务需要 3 步，Agent 循环执行 30 步；
• API Token 消耗暴涨，成本不可控；
• 用户诱导模型绕过规则，执行越权操作；
• Agent 将中间推理过程暴露给用户，泄露内部信息；
• 多工具混用时，参数传错，导致业务数据异常。

因此，构建 Agent 的关键不是“让它越自由越好”，而是：

让 Agent 在清晰边界内，可靠、可控、可观测地完成任务。

三、避坑指南一：不要让 Agent 直接操作高风险接口

这是最重要的一条。

很多人在做 Agent 时，会把真实业务接口直接包装成工具，例如：

• 删除用户；
• 退款；
• 修改订单状态；
• 发送批量短信；
• 发布生产内容；
• 执行 SQL；
• 调用服务器命令；
• 修改权限配置。

这些接口一旦被错误调用，后果可能非常严重。

错误做法

tools = [
    delete_user,
    refund_order,
    execute_sql,
    send_sms
]

然后让模型决定什么时候调用。这种方式风险极高。

推荐做法

对于高风险动作，至少加入以下机制：

1. 只读优先
   初期 Agent 尽量只允许查询，不允许修改。

2. 人工确认
   涉及删除、退款、发布、转账等动作，必须由人工二次确认。

3. 权限校验
   根据用户身份判断是否允许调用某个工具。

4. 参数白名单
   限制工具参数范围，禁止任意输入。

5. 操作审计
   记录用户、时间、工具、参数、返回结果。

6. 幂等保护
   避免重复调用造成重复扣款、重复发消息等。

7. 沙箱环境
   对代码执行、SQL 查询等高风险工具使用隔离环境。

一个更合理的设计是：

def refund_order(order_id: str, amount: float, confirmed: bool = False):
    if not confirmed:
        return {
            "status": "need_confirmation",
            "message": f"请确认是否要为订单 {order_id} 退款 {amount} 元"
        }

    # 真实退款逻辑
    return {
        "status": "success",
        "message": "退款已提交"
    }

Agent 可以提出建议，但不能直接完成高风险操作。

四、避坑指南二：不要相信模型输出的“事实”

大模型非常擅长生成流畅文本，但它可能会产生“幻觉”。尤其在以下场景中：

• 查询业务数据；
• 解释政策规则；
• 生成法律、医疗、金融建议；
• 汇总知识库内容；
• 读取接口结果；
• 分析日志和报表。

如果模型没有真实数据来源，它可能会编造答案。

典型错误

用户问：

我的订单什么时候发货？

模型直接回答：

您的订单预计明天下午发货。

但它根本没有调用订单系统查询。

正确做法

对于事实型问题，应该强制 Agent 调用工具或检索知识库。

例如：

如果用户询问订单、库存、价格、物流、账号状态等事实信息，必须调用对应工具，不允许凭空回答。

并且工具返回什么，最终回答就应该基于什么。

推荐 Prompt 约束

你是一个业务助手。对于任何涉及实时数据、用户数据、订单数据、库存数据的问题，
你必须调用工具查询。若工具没有返回结果，请明确告知用户无法查询到，
不得编造信息。

五、避坑指南三：工具定义要清晰，不要含糊

Agent 是否能正确调用工具，很大程度取决于工具描述是否清楚。

不好的工具描述

{
  "name": "query",
  "description": "查询信息"
}

这种描述过于模糊，模型不知道该查什么，也不知道参数怎么传。

好的工具描述

{
  "name": "get_order_status",
  "description": "根据订单ID查询订单状态、支付状态、发货状态和物流单号。仅用于订单查询，不用于退款或修改订单。",
  "parameters": {
    "order_id": {
      "type": "string",
      "description": "订单ID，例如 202403180001"
    }
  }
}

好的工具描述应该包括：

• 工具用途；
• 什么时候使用；
• 什么时候不要使用；
• 参数类型；
• 参数示例；
• 返回值含义；
• 是否有风险；
• 调用前置条件。

工具越多，描述越要规范，否则 Agent 很容易选错工具。

六、避坑指南四：控制 Agent 的最大执行轮数

Agent 常见架构是 ReAct 模式：

Thought -> Action -> Observation -> Thought -> Action -> Observation -> Final

它会不断思考、调用工具、观察结果。但如果没有限制，就可能陷入死循环。

例如：

1. 调用搜索工具；
2. 搜索结果不满意；
3. 再次搜索；
4. 继续不满意；
5. 再次搜索；
6. 无限循环。

解决方案

设置最大执行步数：

MAX_STEPS = 5

当超过最大步数时，强制停止，并返回当前已知结果。

if step >= MAX_STEPS:
    return "任务执行步数已达到上限，请补充更明确的信息或人工处理。"

这不仅可以避免死循环，还能控制 Token 消耗和接口成本。

七、避坑指南五：必须记录日志和调用链路

Agent 出问题时，如果没有日志，几乎无法排查。

你至少需要记录：

• 用户输入；
• 模型请求；
• 模型输出；
• 工具名称；
• 工具参数；
• 工具返回；
• 执行耗时；
• Token 消耗；
• 异常信息；
• 最终回复。

建议每次任务生成一个 trace_id，把整个调用链串起来。

例如：

import uuid

trace_id = str(uuid.uuid4())
print(f"[TRACE_ID] {trace_id}")

日志不是锦上添花，而是 Agent 生产化的基础能力。

八、避坑指南六：区分“规划”和“执行”

很多 Agent 失败，是因为规划和执行混在一起。

例如用户说：

帮我分析最近一个月销售额下降的原因，并给出优化建议。

这个任务可能需要：

1. 查询最近一个月销售数据；
2. 查询上个月销售数据；
3. 查询商品维度数据；
4. 查询渠道维度数据；
5. 查询用户转化率；
6. 对比变化；
7. 生成结论。

如果 Agent 一上来就回答，很容易泛泛而谈。

更好的方式是先规划：

我将按以下步骤分析：
1. 获取最近一个月销售数据；
2. 获取对比周期销售数据；
3. 分析商品、渠道、用户转化三个维度；
4. 总结下降原因；
5. 给出优化建议。

然后逐步调用工具执行。

对于复杂任务，建议采用：

• Plan-and-Execute；
• ReAct；
• Workflow + LLM；
• 多 Agent 分工；
• 状态机。

不过在真实业务中，工作流 + LLM 往往比完全自主 Agent 更稳定。

九、避坑指南七：不要把所有事情都交给 Agent

Agent 不是万能的。很多场景其实并不适合完全用 Agent。

适合 Agent 的场景

• 多步骤任务；
• 需要调用多个工具；
• 用户输入不固定；
• 需要自然语言交互；
• 需要一定推理和判断；
• 流程有一定弹性。

不适合 Agent 的场景

• 强确定性的固定流程；
• 高并发低延迟接口；
• 严格合规审批；
• 核心交易链路；
• 不能容忍错误的生产操作；
• 规则非常明确的业务判断。

如果一个任务用普通代码、规则引擎、工作流系统就能稳定完成，就不一定需要 Agent。

正确思路是：

用确定性代码处理确定性逻辑，用大模型处理非结构化理解、推理和生成。

十、避坑指南八：上下文不要无限堆叠

很多人做 Agent 时，会把所有历史对话、所有文档、所有工具结果都塞进上下文。这会带来几个问题：

• Token 成本升高；
• 模型注意力分散；
• 关键信息被淹没；
• 超出上下文长度；
• 历史错误影响当前回答。

推荐做法

1. 短期记忆
   保留最近几轮对话。

2. 长期记忆
   存储结构化用户偏好，例如语言、角色、常用项目。

3. 知识检索
   用向量检索或关键词检索，只取相关文档片段。

4. 上下文压缩
   将历史信息总结成摘要。

5. 任务隔离
   不同任务使用不同上下文，避免污染。

示例：

def summarize_history(messages):
    # 实际项目中可调用 LLM 对历史对话进行摘要
    return "用户正在咨询订单相关问题，关注发货时间和物流状态。"

十一、避坑指南九：输出格式要结构化

如果 Agent 的输出要给后端系统继续处理，千万不要只让它输出自然语言。

例如你希望它返回操作决策，就应该使用 JSON：

{
  "action": "query_order",
  "order_id": "202403180001",
  "need_human_confirm": false
}

而不是：

我觉得可以查询一下订单 202403180001。

结构化输出可以带来：

• 更容易解析；
• 更容易校验；
• 更容易测试；
• 更容易接入工作流；
• 更容易控制风险。

在工程中，建议使用 JSON Schema、Pydantic 或函数调用规范来限制输出。

十二、避坑指南十：要有失败兜底策略

Agent 不可能每次都成功。你需要提前设计失败处理。

常见失败包括：

• 模型返回格式错误；
• 工具调用失败；
• API 超时；
• 用户输入不完整；
• 查询结果为空；
• 模型不确定；
• 超出最大执行步数；
• 安全策略拦截。

推荐兜底方式：

1. 请求用户补充信息；
2. 返回已知部分结果；
3. 转人工处理；
4. 降级为普通问答；
5. 重试一次但不无限重试；
6. 记录异常，进入监控告警。

例如：

try:
    result = tool(**arguments)
except Exception as e:
    return {
        "status": "error",
        "message": "工具调用失败，请稍后重试或联系人工客服。",
        "error": str(e)
    }

十三、AI Agent 简化版源码示例

下面给出一个简化版 Python Agent 示例，用来演示：

• 工具注册；
• 工具调用；
• 最大步数控制；
• 安全确认；
• 日志追踪；
• 简单意图判断；
• 最终结果返回。

说明：为了方便理解，下面示例没有真实接入大模型，而是用规则模拟 LLM 的决策过程。实际项目中，你可以把 mock_llm_decide 替换成 OpenAI、Claude、通义千问、智谱、DeepSeek 等模型接口。

1. 完整源码

import uuid
import json
from typing import Callable, Dict, Any, List


class Tool:
    """
    工具定义类
    """

    def __init__(
        self,
        name: str,
        description: str,
        func: Callable,
        risk_level: str = "low"
    ):
        self.name = name
        self.description = description
        self.func = func
        self.risk_level = risk_level

    def run(self, **kwargs):
        return self.func(**kwargs)


class Agent:
    """
    一个简化版 AI Agent
    """

    def __init__(self, tools: List[Tool], max_steps: int = 5):
        self.tools = {tool.name: tool for tool in tools}
        self.max_steps = max_steps

    def log(self, trace_id: str, message: str, data: Any = None):
        print(f"[TRACE_ID={trace_id}] {message}")
        if data is not None:
            print(json.dumps(data, ensure_ascii=False, indent=2))

    def run(self, user_input: str):
        trace_id = str(uuid.uuid4())

        self.log(trace_id, "收到用户输入", {
            "user_input": user_input
        })

        context = {
            "user_input": user_input,
            "observations": []
        }

        for step in range(1, self.max_steps + 1):
            self.log(trace_id, f"开始第 {step} 步决策")

            decision = self.mock_llm_decide(context)

            self.log(trace_id, "模型决策结果", decision)

            if decision["type"] == "final":
                self.log(trace_id, "任务完成")
                return decision["answer"]

            if decision["type"] == "tool_call":
                tool_name = decision["tool_name"]
                arguments = decision.get("arguments", {})

                if tool_name not in self.tools:
                    error_msg = f"工具 {tool_name} 不存在"
                    self.log(trace_id, error_msg)
                    return error_msg

                tool = self.tools[tool_name]

                # 高风险工具必须确认
                if tool.risk_level == "high" and not arguments.get("confirmed"):
                    confirm_msg = {
                        "status": "need_confirmation",
                        "message": f"操作 {tool_name} 属于高风险操作，请用户确认后再执行。",
                        "arguments": arguments
                    }
                    self.log(trace_id, "高风险操作被拦截", confirm_msg)
                    return confirm_msg["message"]

                try:
                    result = tool.run(**arguments)
                    self.log(trace_id, f"工具 {tool_name} 执行结果", result)

                    context["observations"].append({
                        "tool_name": tool_name,
                        "arguments": arguments,
                        "result": result
                    })

                except Exception as e:
                    self.log(trace_id, "工具调用异常", {
                        "tool_name": tool_name,
                        "error": str(e)
                    })
                    return "工具调用失败，请稍后重试或联系人工处理。"

        self.log(trace_id, "达到最大执行步数")
        return "任务执行步数已达到上限，请补充更明确的信息或转人工处理。"

    def mock_llm_decide(self, context: Dict[str, Any]):
        """
        用规则模拟 LLM 决策。
        实际项目中可以替换为大模型函数调用。
        """

        user_input = context["user_input"]
        observations = context["observations"]

        # 如果已经有工具结果，则生成最终回答
        if observations:
            last = observations[-1]
            tool_name = last["tool_name"]
            result = last["result"]

            if tool_name == "get_order_status":
                return {
                    "type": "final",
                    "answer": (
                        f"查询结果：订单状态为 {result['order_status']}，"
                        f"支付状态为 {result['payment_status']}，"
                        f"发货状态为 {result['shipping_status']}，"
                        f"物流单号为 {result.get('tracking_no', '暂无')}。"
                    )
                }

            if tool_name == "search_knowledge_base":
                return {
                    "type": "final",
                    "answer": f"根据知识库查询结果：{result['content']}"
                }

            if tool_name == "refund_order":
                return {
                    "type": "final",
                    "answer": result["message"]
                }

        # 简单意图识别：订单查询
        if "订单" in user_input and ("状态" in user_input or "发货" in user_input):
            order_id = self.extract_order_id(user_input)
            if not order_id:
                return {
                    "type": "final",
                    "answer": "请提供订单ID，我才能帮你查询订单状态。"
                }

            return {
                "type": "tool_call",
                "tool_name": "get_order_status",
                "arguments": {
                    "order_id": order_id
                }
            }

        # 简单意图识别：退款
        if "退款" in user_input:
            order_id = self.extract_order_id(user_input)
            return {
                "type": "tool_call",
                "tool_name": "refund_order",
                "arguments": {
                    "order_id": order_id or "unknown",
                    "amount": 99.0,
                    "confirmed": False
                }
            }

        # 简单意图识别：知识库查询
        if "怎么" in user_input or "如何" in user_input:
            return {
                "type": "tool_call",
                "tool_name": "search_knowledge_base",
                "arguments": {
                    "query": user_input
                }
            }

        return {
            "type": "final",
            "answer": "我暂时无法理解你的需求，请换一种方式描述。"
        }

    def extract_order_id(self, text: str):
        """
        简单提取订单ID：真实项目建议用更可靠的实体识别或正则规则。
        """
        import re
        match = re.search(r"\d{6,}", text)
        if match:
            return match.group(0)
        return None


def get_order_status(order_id: str):
    """
    模拟订单查询工具
    """
    fake_db = {
        "202403180001": {
            "order_status": "已支付",
            "payment_status": "支付成功",
            "shipping_status": "已发货",
            "tracking_no": "SF123456789"
        },
        "202403180002": {
            "order_status": "已支付",
            "payment_status": "支付成功",
            "shipping_status": "待发货",
            "tracking_no": None
        }
    }

    if order_id not in fake_db:
        return {
            "order_status": "未知",
            "payment_status": "未知",
            "shipping_status": "未查询到订单",
            "tracking_no": None
        }

    return fake_db[order_id]


def search_knowledge_base(query: str):
    """
    模拟知识库查询工具
    """
    docs = {
        "如何申请发票": "你可以在订单详情页点击“申请发票”，填写发票抬头和邮箱后提交。",
        "怎么修改地址": "如果订单尚未发货，可以在订单详情页修改收货地址；已发货订单无法修改。"
    }

    for key, value in docs.items():
        if key in query:
            return {
                "content": value
            }

    return {
        "content": "知识库中没有找到完全匹配的内容，建议联系人工客服。"
    }


def refund_order(order_id: str, amount: float, confirmed: bool = False):
    """
    模拟退款工具：高风险操作，需要确认
    """
    if not confirmed:
        return {
            "status": "need_confirmation",
            "message": f"请确认是否要为订单 {order_id} 退款 {amount} 元。"
        }

    return {
        "status": "success",
        "message": f"订单 {order_id} 已提交退款申请，金额 {amount} 元。"
    }


if __name__ == "__main__":
    tools = [
        Tool(
            name="get_order_status",
            description="根据订单ID查询订单状态、支付状态、发货状态和物流单号。",
            func=get_order_status,
            risk_level="low"
        ),
        Tool(
            name="search_knowledge_base",
            description="查询售后、发票、地址修改等常见问题知识库。",
            func=search_knowledge_base,
            risk_level="low"
        ),
        Tool(
            name="refund_order",
            description="为指定订单发起退款申请。该操作属于高风险操作，必须用户确认。",
            func=refund_order,
            risk_level="high"
        )
    ]

    agent = Agent(tools=tools, max_steps=5)

    print("\n--- 示例1：订单查询 ---")
    print(agent.run("帮我查一下订单 202403180001 的发货状态"))

    print("\n--- 示例2：知识库查询 ---")
    print(agent.run("如何申请发票？"))

    print("\n--- 示例3：高风险退款 ---")
    print(agent.run("帮我把订单 202403180002 退款"))

2. 运行结果示例

运行后你可能会看到类似输出：

--- 示例1：订单查询 ---
查询结果：订单状态为 已支付，支付状态为 支付成功，发货状态为 已发货，物流单号为 SF123456789。

--- 示例2：知识库查询 ---
根据知识库查询结果：你可以在订单详情页点击“申请发票”，填写发票抬头和邮箱后提交。

--- 示例3：高风险退款 ---
操作 refund_order 属于高风险操作，请用户确认后再执行。

这个示例虽然简单，但已经包含了 Agent 的几个关键设计点：

• 工具不是随便调用，而是统一注册；
• 高风险工具有风险等级；
• 执行过程有最大步数；
• 每次任务有 trace_id；
• 工具调用异常有兜底；
• 事实查询依赖工具结果；
• 不确定时要求用户补充信息。

十四、真实项目中的推荐架构

如果你要把 Agent 用到真实业务中，可以参考以下架构：

用户输入
  ↓
权限校验
  ↓
安全过滤
  ↓
意图识别
  ↓
任务规划
  ↓
工具选择
  ↓
参数校验
  ↓
工具执行
  ↓
结果校验
  ↓
LLM 总结
  ↓
输出给用户
  ↓
日志审计 / 监控告警

其中比较关键的是：

1. 权限校验

不是所有用户都能调用所有工具。例如普通用户不能查看他人订单，客服不能直接退款超过一定金额，运营不能直接发布高风险内容。

2. 参数校验

模型生成的参数必须校验，不能直接信任。

例如：

if amount <= 0 or amount > 1000:
    raise ValueError("退款金额不合法")

3. 工具结果校验

工具返回异常或空结果时，模型不能编造答案。

4. 人工确认

高风险动作必须人工确认，必要时进入审批流。

5. 监控告警

关注以下指标：

• Agent 成功率；
• 工具调用失败率；
• 平均响应时间；
• 平均执行步数；
• Token 消耗；
• 用户满意度；
• 人工接管率；
• 高风险操作拦截次数。

十五、AI Agent 落地检查清单

上线前建议逐项检查：

• [ ] 是否限制了 Agent 可调用的工具？
• [ ] 是否区分了低风险和高风险工具？
• [ ] 高风险操作是否需要人工确认？
• [ ] 是否设置最大执行步数？
• [ ] 是否有完整日志和 trace_id？
• [ ] 是否有权限校验？
• [ ] 是否有参数校验？
• [ ] 是否有失败兜底？
• [ ] 是否禁止模型编造事实数据？
• [ ] 是否使用结构化输出？
• [ ] 是否对上下文进行了长度控制？
• [ ] 是否有测试集评估 Agent 表现？
• [ ] 是否监控 Token 成本？
• [ ] 是否支持转人工？
• [ ] 是否对工具调用结果进行审计？

十六、总结

AI Agent 的价值不在于让大模型“自由发挥”，而在于让它在工程约束下完成复杂任务。

如果只是做 Demo，Agent 可以很炫；但如果要进入生产环境，就必须重视安全、权限、工具、日志、兜底、成本和稳定性。

最核心的原则可以总结为五句话：

1. 能不用 Agent，就不要强行用 Agent。
2. 能用确定性代码解决的，不要交给模型猜。
3. 涉及事实数据，必须调用真实工具或检索系统。
4. 涉及高风险动作，必须人工确认和权限控制。
5. Agent 必须可观测、可审计、可回滚、可降级。

AI Agent 不是一个单独的 Prompt，也不是简单的工具调用，而是一套完整的工程系统。只有把大模型能力和软件工程约束结合起来，才能真正做出可靠、可用、可上线的智能体应用。