第 5 章 · Agent 智能体

本章目标：让模型从"只动嘴"变成"能动手"——自己决定调哪个接口、查哪张表、执行什么操作。

5.1 什么是 Agent

前几章的模型都只会"说话"：你给它文字，它回文字。但现实任务里，很多事模型做不到，必须靠代码：

• 查数据库、调外部接口、发邮件、操作文件……

Agent（智能体）的核心思想：把"模型"和"工具"结合起来，让模型在需要时主动调用工具，拿到结果后继续思考，直到完成任务。

它的最小结构就三个部件：

LLM（大脑） + 工具（手脚） + 循环（让大脑和手脚反复配合）

打个比方：前几章的模型是个"只会说话的顾问"；Agent 是个"能打电话、能查系统、能动手干活的助手"。它不只是给你建议，它真的把事办了。

5.2 Function Calling：Agent 的地基

Function Calling（函数调用，也叫 Tool Use）是 Agent 的地基。它的机制是：

模型不再只输出文字，而是输出一段结构化 JSON，告诉你"我想调用这个函数、参数是这个"。你的代码去执行这个函数，把结果喂回模型，模型再继续。

整个过程是一个"多轮"的对话，但中间穿插了"调用工具"的动作。

完整代码：一个能查数据库的 Agent

import os
from openai import OpenAI
import sqlite3, json

client = OpenAI(api_key=os.environ.get("DEEPSEEK_API_KEY", "sk-你的key"), base_url="https://api.deepseek.com")

# 准备一个数据库
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE stock (name TEXT, qty INTEGER)")
conn.executemany("INSERT INTO stock VALUES (?,?)",
                 [("玉米", 5000), ("小麦", 3000), ("水稻", 2000)])
conn.commit()

# 1. 定义工具的 JSON 描述（告诉模型"有这些工具可用"）
tools = [
    {
        "type": "function",
        "function": {
            "name": "query_stock",
            "description": "查询某个粮食品种的库存数量（单位：公斤）",
            "parameters": {
                "type": "object",
                "properties": {
                    "grain_name": {
                        "type": "string",
                        "description": "粮食品种名称，如 玉米、小麦、水稻"
                    }
                },
                "required": ["grain_name"],
            },
        },
    }
]

# 2. 工具的真实实现（模型只是"说"要调，真正执行的是这里）
def query_stock(grain_name: str) -> str:
    row = conn.execute("SELECT qty FROM stock WHERE name=?", (grain_name,)).fetchone()
    return f"{grain_name} 的库存是 {row[0]} 公斤" if row else f"未找到 {grain_name} 的库存"

TOOL_MAP = {"query_stock": query_stock}

def run_agent(user_question: str):
    messages = [{"role": "user", "content": user_question}]

    # 第一轮：问模型要不要调工具
    resp = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=messages,
        tools=tools,
    )
    msg = resp.choices[0].message

    # 如果模型想调工具
    if msg.tool_calls:
        # 把模型的"想法"记录进对话
        messages.append(msg)

        for call in msg.tool_calls:
            fn_name = call.function.name
            fn_args = json.loads(call.function.arguments)   # 模型给的是 JSON 字符串
            print(f"[调用工具] {fn_name}({fn_args})")

            result = TOOL_MAP[fn_name](**fn_args)          # 真正执行工具

            # 把工具执行结果作为 tool 消息喂回去
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })

        # 第二轮：模型拿到工具结果，生成最终回答
        resp2 = client.chat.completions.create(
            model="deepseek-v4-flash",
            messages=messages,
        )
        return resp2.choices[0].message.content

    # 如果模型不需要工具，直接回答
    return msg.content

print(run_agent("玉米的库存还有多少？"))

这段代码是 Agent 的核心，请一定吃透这几步

1. tools 参数：用 JSON 描述工具（名字、用途、参数、哪些必填）。模型就是靠这段描述"看懂"工具有什么用、该怎么调。
2. 模型返回 msg.tool_calls：它不执行任何东西，只是"表达意愿"——"我想调 query_stock，参数是 玉米"。
3. 你的代码真正执行：TOOL_MAP[fn_name](**fn_args) 才是真正查库的地方。
4. 结果喂回去：以 role: "tool" 的身份，把结果连同一个 tool_call_id 塞回对话。
5. 第二轮调用：模型拿到结果，组织成最终的人话回答。

关键认知：模型是"决策者"，代码是"执行者"。 模型永远不碰你的数据库、不碰你的接口，它只负责"决定调什么、传什么参"。这意味着你的工具实现可以做权限校验、日志、限流——安全边界始终掌握在你手里。这也是为什么说 Agent 开发其实是工程活。

5.3 ReAct：推理与行动交替的循环

一个工具太简单。真实任务往往要"多步 + 多工具"：先查 A，根据 A 的结果决定查 B，最后综合。这需要让模型进入一个循环，这就是 ReAct（Reasoning + Acting，推理 + 行动）：

思考（Thought）→ 行动（Action，调工具）→ 观察（Observation，工具结果）→ 再思考 → ... → 最终回答

把 4.2 的单轮改成循环版（最多迭代 N 次，防止死循环）：

def run_agent_loop(user_question: str, max_steps=5):
    messages = [{"role": "user", "content": user_question}]

    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="deepseek-v4-flash",
            messages=messages,
            tools=tools,
        )
        msg = resp.choices[0].message

        if msg.tool_calls:                    # 还想调工具 → 执行，继续循环
            messages.append(msg)
            for call in msg.tool_calls:
                fn_name = call.function.name
                fn_args = json.loads(call.function.arguments)
                result = TOOL_MAP[fn_name](**fn_args)
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": result,
                })
            continue                          # 进入下一轮
        else:                                  # 不再调工具 → 给出最终答案
            return msg.content

    return "达到最大步骤数，任务未完成。"      # 兜底，防死循环

这个循环版就能处理"查多个品种的库存，再汇总对比"这类多步任务了。

5.4 完整多工具 Agent 示例

给 Agent 配上多个工具，它就能"查库存 + 查订单 + 做数学换算"：

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_stock",
            "description": "查询某粮食品种的库存（公斤）",
            "parameters": {
                "type": "object",
                "properties": {"grain_name": {"type": "string", "description": "粮食品种"}},
                "required": ["grain_name"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calc_fold",
            "description": "含水率折算：折合重量 = 实测重量×(1-实测含水率)/(1-标准含水率)",
            "parameters": {
                "type": "object",
                "properties": {
                    "actual_weight": {"type": "number", "description": "实测重量，公斤"},
                    "actual_moisture": {"type": "number", "description": "实测含水率，如 0.14"},
                    "std_moisture": {"type": "number", "description": "标准含水率，如 0.12"},
                },
                "required": ["actual_weight", "actual_moisture", "std_moisture"],
            },
        },
    },
]

def query_stock(grain_name):
    row = conn.execute("SELECT qty FROM stock WHERE name=?", (grain_name,)).fetchone()
    return f"{grain_name} 库存 {row[0]} 公斤" if row else f"无 {grain_name} 库存"

def calc_fold(actual_weight, actual_moisture, std_moisture):
    folded = actual_weight * (1 - actual_moisture) / (1 - std_moisture)
    return f"折合重量为 {folded:.1f} 公斤"

TOOL_MAP = {"query_stock": query_stock, "calc_fold": calc_fold}

# 这个任务需要先查库存、再做折算、最后给结论，模型会自己决定调用的顺序
print(run_agent_loop("玉米实际入库 5000 公斤，含水率 14%，按标准 12% 折算后是多少？并告诉我当前玉米库存。"))

观察点：模型会先调 calc_fold 算折算，再调 query_stock 查库存，最后综合成一段话。它"自己"决定了工具的调用顺序——这就是 Agent 和"写死的流程"的本质区别。

5.5 用框架写 Agent（以 CrewAI 为例）

手写循环能帮你理解原理，但生产上更多用框架。2026 年主流框架的定位：

建议路线：先用 CrewAI 把"多 Agent"概念吃透（它最直观），需要精细控制时再上 LangGraph，需要快速落地/可视化时用 Dify。

CrewAI 入门示例（感受一下"角色分工"）：

# pip install crewai
from crewai import Agent, Task, Crew, LLM

llm = LLM(model="deepseek/deepseek-v4-flash")   # CrewAI 可接 DeepSeek

研究员 = Agent(
    role="数据分析师",
    goal="从原始数据里提炼关键结论",
    backstory="你在智粮系统工作 10 年，擅长从入库数据里发现问题",
    llm=llm,
)
写手 = Agent(
    role="汇报写手",
    goal="把分析结论写成简洁的周报",
    backstory="你擅长把复杂数据写成领导一眼能看懂的话",
    llm=llm,
)

分析任务 = Task(description="分析本周各品种入库量，找出异常", agent=研究员)
写作任务 = Task(description="基于分析结果写一份 200 字周报", agent=写手)

crew = Crew(agents=[研究员, 写手], tasks=[分析任务, 写作任务])
result = crew.kickoff()
print(result)

看到区别了吗：手写版是"一个模型在一个循环里调工具"；CrewAI 是"多个有身份的 Agent 分工协作"。先手写理解本质，再用框架提效，这个顺序别颠倒。

5.6 Agent 的稳定性：最难的部分，也是你的主场

Agent 容易出的问题，比普通对话多得多：

1. 死循环：模型一直调同一个工具，转不出来（所以要有 max_steps 兜底）。
2. 乱调工具：参数传错、调了不该调的接口。
3. 中途跑偏：多步任务做到一半忘了目标。
4. 编造结果：工具返回了 A，模型却"脑补"成 B。

对应的工程化手段（这些全是传统工程师的拿手活）：

再强调一次本章最重要的判断：Agent 最难的不是"能不能跑起来"，是"跑得稳不稳"。 一个 Demo 半天就能跑通，但要让它在生产环境稳定服务、出了问题能定位，靠的是工程能力——评测、兜底、日志、监控。这些，正是你有十几年经验沉淀的地方，是纯算法背景的人补不上的短板。

5.7 本章小结与练习

你该记住的：

• Agent = LLM（决策）+ 工具（执行）+ 循环（协作）。
• Function Calling：模型输出"想调哪个函数、什么参数"的 JSON，你的代码执行，结果喂回去。
• 模型永远是"决策者"，代码才是"执行者"，安全边界在你手里。
• ReAct 循环 = 思考 → 行动 → 观察，反复直到完成。
• 框架先 CrewAI 上手，再按需选 LangGraph / Dify / Coze。
• 稳定性是 Agent 最大的坑，也是你的主场。

练习：

1. 把 4.2 的单轮 Agent 跑通，再加一个工具（比如 add_record 往库存表插入一条记录），让它"先查再增"。
2. 给 run_agent_loop 加"重复调用检测"：如果连续两次调用同一个工具且参数相同，直接终止。
3. 用 CrewAI 把 4.5 的示例跑起来，感受多 Agent 协作。

下一章：把前面所有东西拼成"能上线的产品"——记忆管理、评测、成本控制、部署，以及多 Agent / MCP / 微调这些进阶方向。