大厂老兵从零学 Agent 开发：从概念到可运行实战

m0_73571447

287人浏览 · 2026-11-05 20:33:36

m0_73571447 · 2026-11-05 20:33:36 发布

1. 写在前面：为什么大厂老兵也要重新学

如果你在大厂做过几年后端、算法或前端，你一定有过这种错觉：Agent 不就是「LLM + 几个 if-else + 调几个 API」吗？

真正动手之后你会发现，坑远比想象的多：

模型会幻觉调用不存在的工具；
多轮对话里上下文会爆炸；
工具调用失败后没有重试与降级；
上线后无法观测每一步到底发生了什么；
成本、延迟、稳定性三座大山压下来，Demo 和生产的差距是数量级的。

这篇文章面向「有工程基础、但没系统做过 Agent」的开发者，用一个能跑起来的完整项目串起 Agent 开发的核心链路：工具调用、ReAct 循环、记忆管理、多 Agent 协作、可观测性与工程化落地。

阅读前提：会 Python，了解 HTTP 与 JSON，知道什么是 Prompt。不需要你有任何 Agent 框架经验。

2. Agent 到底是什么：和普通 LLM 调用的区别

先给一个工程视角的定义：

Agent = LLM（决策大脑） + Tools（手脚） + Memory（记忆） + Loop（循环）

普通 LLM 调用是一次性的：输入 Prompt，输出文本，结束。

Agent 是闭环的：模型输出「我要调用某个工具」，程序执行工具，把结果回灌给模型，模型再决定下一步，直到任务完成或达到终止条件。

这个循环就是 Agent 的心脏。后面所有的复杂度——记忆、多 Agent、可观测性——都是围绕这个循环长出来的。

2.1 三种常见形态

形态	特点	典型场景
Workflow（工作流）	步骤固定，LLM 只做单点判断	文档摘要、分类打标
Agent（单智能体）	模型自主决定调用哪些工具、调几次	问答助手、代码修复
Multi-Agent（多智能体）	多个角色分工协作	复杂研究、长链路任务

工程建议：能用 Workflow 解决的，不要上 Agent。Agent 的自由度带来的是不可控性，先问自己「步骤是否真的无法预先确定」。

3. 环境准备

本文使用 Python 3.10+，依赖尽量精简，方便你看清每一步在做什么。

python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate

pip install openai pydantic python-dotenv tenacity

.env 文件：

OPENAI_API_KEY=sk-xxxx
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-4o-mini

如果你用的是国内兼容 OpenAI 协议的模型服务，只需要改 OPENAI_BASE_URL 和 MODEL_NAME，本文代码无需改动。

统一封装一个客户端：

# llm.py
import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

MODEL = os.getenv("MODEL_NAME", "gpt-4o-mini")


def chat(messages, tools=None, temperature=0.0):
    """统一的对话入口，tools 为空时退化为普通对话。"""
    kwargs = {
        "model": MODEL,
        "messages": messages,
        "temperature": temperature,
    }
    if tools:
        kwargs["tools"] = tools
        kwargs["tool_choice"] = "auto"
    return client.chat.completions.create(**kwargs)

4. 第一步：让模型学会调用工具

工具调用的本质是：用 JSON Schema 描述函数，模型输出结构化的调用意图，你来执行。

4.1 定义工具

# tools.py
import json
import math
from datetime import datetime

# 1) 工具的真实实现
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def calculate(expression: str) -> str:
    """只允许数学表达式，避免 eval 注入。"""
    allowed = set("0123456789+-*/(). %")
    if not set(expression) <= allowed:
        return "错误：表达式包含非法字符"
    try:
        return str(eval(expression, {"__builtins__": {}}, {"math": math}))
    except Exception as e:
        return f"计算失败：{e}"


def search_docs(query: str) -> str:
    """模拟一个知识库检索工具。"""
    fake_db = {
        "退款": "退款需在订单完成后 7 天内申请，原路退回。",
        "发票": "电子发票可在订单详情页自助开具，1 个工作日内发送。",
    }
    for key, value in fake_db.items():
        if key in query:
            return value
    return "未检索到相关内容。"


# 2) 工具注册表：名字 -> 函数
TOOL_REGISTRY = {
    "get_current_time": get_current_time,
    "calculate": calculate,
    "search_docs": search_docs,
}

# 3) 给模型看的 Schema
TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前时间，当用户询问现在几点、今天日期时使用",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone": {"type": "string", "description": "时区，默认 Asia/Shanghai"}
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算，支持加减乘除与括号",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式，如 (12+8)*3"}
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_docs",
            "description": "检索公司内部知识库，回答退款、发票等业务问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "检索关键词"}
                },
                "required": ["query"],
            },
        },
    },
]

几个容易踩的坑：

description 是模型选择工具的唯一依据，写得越清楚，误调用越少；
required 不要乱填，非必填参数填进去会逼模型编造值；
工具实现里一定要做输入校验，模型给的参数不可信。

4.2 执行工具调用

# executor.py
import json
from tools import TOOL_REGISTRY


def execute_tool_call(tool_call) -> str:
    name = tool_call.function.name
    raw_args = tool_call.function.arguments or "{}"

    if name not in TOOL_REGISTRY:
        return json.dumps({"error": f"未知工具：{name}"}, ensure_ascii=False)

    try:
        args = json.loads(raw_args)
    except json.JSONDecodeError:
        return json.dumps({"error": "参数不是合法 JSON"}, ensure_ascii=False)

    try:
        result = TOOL_REGISTRY[name](**args)
        return str(result)
    except TypeError as e:
        return json.dumps({"error": f"参数不匹配：{e}"}, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"error": f"工具执行异常：{e}"}, ensure_ascii=False)

注意这里所有异常都被转成字符串返回给模型，而不是抛出去。因为模型需要看到错误信息才能自我纠正，直接崩溃会让整个循环中断。

5. 第二步：实现 ReAct 循环

ReAct（Reasoning + Acting）是最经典的 Agent 范式：模型先思考，再行动，观察结果，继续思考。

# agent.py
import json
from llm import chat
from tools import TOOL_SCHEMAS
from executor import execute_tool_call

SYSTEM_PROMPT = """你是一个严谨的助手。
规则：
1. 需要实时信息或计算时，必须调用工具，不要凭空编造。
2. 工具返回错误时，分析原因并尝试修正参数，最多重试 2 次。
3. 信息足够时直接给出最终答案，不要无意义地反复调用工具。
"""


def run_agent(user_input: str, max_steps: int = 8, verbose: bool = True) -> str:
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_input},
    ]

    for step in range(max_steps):
        resp = chat(messages, tools=TOOL_SCHEMAS)
        msg = resp.choices[0].message

        # 没有工具调用 -> 任务结束
        if not msg.tool_calls:
            return msg.content or ""

        # 把模型的决策写回上下文（关键！）
        messages.append({
            "role": "assistant",
            "content": msg.content,
            "tool_calls": [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {
                        "name": tc.function.name,
                        "arguments": tc.function.arguments,
                    },
                }
                for tc in msg.tool_calls
            ],
        })

        # 逐个执行工具并回灌结果
        for tc in msg.tool_calls:
            result = execute_tool_call(tc)
            if verbose:
                print(f"[step {step}] {tc.function.name}({tc.function.arguments}) -> {result}")
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result,
            })

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

跑一下：

if __name__ == "__main__":
    print(run_agent("帮我算一下 (128 + 72) * 3 是多少，再告诉我现在几点"))

预期输出类似：

[step 0] calculate({"expression": "(128+72)*3"}) -> 600
[step 0] get_current_time({}) -> 2026-11-05 20:32:49
最终答案：计算结果是 600，当前时间是 2026-11-05 20:32:49。

5.1 三个必须加的护栏

护栏一：最大步数。上面已经加了 max_steps，防止模型陷入死循环烧钱。

护栏二：工具调用去重。同一个工具用相同参数连续调用，大概率是模型卡住了：

def is_repeated(messages, name, args, window=4):
    """检查最近 window 条消息里是否出现过完全相同的调用。"""
    recent = [m for m in messages[-window:] if m.get("role") == "assistant"]
    for m in recent:
        for tc in m.get("tool_calls") or []:
            if tc["function"]["name"] == name and tc["function"]["arguments"] == args:
                return True
    return False

护栏三：超时与重试。工具调用可能卡住，用 tenacity 包一层：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=8),
    retry=retry_if_exception_type((TimeoutError, ConnectionError)),
)
def safe_call(fn, **kwargs):
    return fn(**kwargs)

6. 第三步：记忆管理

Agent 跑几轮之后，messages 会越来越长，最终撞上上下文窗口。工程上有三种处理方式。

6.1 滑动窗口

最简单，只保留最近 N 轮：

def trim_messages(messages, max_tokens=6000, keep_system=True):
    """粗略按字符数裁剪，保留 system 与最近消息。"""
    system = [m for m in messages if m["role"] == "system"] if keep_system else []
    rest = [m for m in messages if m["role"] != "system"]

    total, kept = 0, []
    for m in reversed(rest):
        length = len(str(m.get("content") or "")) + 200  # 预留 tool_calls 开销
        if total + length > max_tokens:
            break
        kept.append(m)
        total += length
    return system + list(reversed(kept))

6.2 摘要压缩

把早期对话压缩成一段摘要，保留语义：

def summarize(messages):
    history = "\n".join(
        f"{m['role']}: {m.get('content') or ''}" for m in messages if m["role"] != "system"
    )
    resp = chat([
        {"role": "system", "content": "把下面的对话压缩成 200 字以内的要点，保留关键事实与结论。"},
        {"role": "user", "content": history},
    ])
    return resp.choices[0].message.content

6.3 外部记忆（向量检索）

长期记忆放向量库，按需召回。生产环境推荐这个方案，但要注意召回质量直接决定回答质量，别把脏数据塞进去。

方案	优点	缺点	适用
滑动窗口	实现简单、零成本	丢失早期信息	短会话
摘要压缩	保留语义	有信息损耗、多一次调用	中等长度会话
向量检索	可无限扩展	工程复杂、召回不稳	长期记忆

7. 第四步：多 Agent 协作

当任务复杂到单个 Agent 的 Prompt 装不下时，拆成多个角色。

一个极简实现：

def planner(task: str) -> str:
    resp = chat([
        {"role": "system", "content": "你是任务规划专家，把任务拆成 3-5 个可执行子步骤，每行一个。"},
        {"role": "user", "content": task},
    ])
    return resp.choices[0].message.content


def worker(role: str, instruction: str, context: str = "") -> str:
    resp = chat([
        {"role": "system", "content": f"你是{role}，只完成分配给你的部分，输出简洁专业。"},
        {"role": "user", "content": f"任务：{instruction}\n参考资料：{context}"},
    ])
    return resp.choices[0].message.content


def multi_agent(task: str) -> str:
    plan = planner(task)
    research = worker("研究员", f"围绕以下计划收集要点：\n{plan}")
    draft = worker("撰稿人", f"根据计划与资料写初稿：\n{plan}", research)
    review = worker("审核员", f"审核初稿并给出修改意见：\n{draft}")
    final = worker("撰稿人", f"根据审核意见修订：\n{review}", draft)
    return final

关键经验：多 Agent 的收益来自上下文隔离，而不是「人多力量大」。每个 Agent 只看到自己需要的信息，反而比一个巨型 Prompt 更稳定。但代价是调用次数翻倍，成本要算清楚。

8. 第五步：可观测性与工程化

Demo 能跑不代表能上线。生产环境至少要具备：

8.1 结构化日志

import logging
import json
import time

logger = logging.getLogger("agent")
logger.setLevel(logging.INFO)


def log_step(trace_id, step, event, payload):
    logger.info(json.dumps({
        "trace_id": trace_id,
        "step": step,
        "event": event,
        "payload": payload,
        "ts": time.time(),
    }, ensure_ascii=False))

在 run_agent 里埋点：每次模型调用记录 token 数与耗时，每次工具调用记录入参与结果。出问题时能一眼定位是「模型决策错」还是「工具执行错」。

8.2 关键指标

指标	说明	告警阈值参考
平均步数	单任务工具调用轮次	> 6 需排查
工具失败率	工具返回 error 的占比	> 10%
P95 延迟	端到端耗时	按业务定
单任务成本	token 消耗折算	按预算定