这次我们来看一个很典型的工程向主题:使用 Python 构建真实 AI 代理的 Agentic AI Engineering。注意标题里的三个关键词:真实、AI 代理、工程。也就是说,这本书/课程不是给你讲大模型 API 怎么调,也不是给你看几个 ChatBot Demo,而是要把 AI 代理当作一个可落地、可维护、可上线的软件工程系统来处理。
如果你已经写过不少 Python 脚本,也在 LangChain、LangGraph 或 OpenAI Function Calling 上做过一些小实验,但总觉得“代理”这个东西只停留在概念阶段,跑起来容易飘,遇到工具调用失败、上下文爆炸、多轮状态错乱就不知道怎么收场,那这篇文章就是你需要的。
这篇文章会把 Agentic AI 工程这条线拆开讲:先说明 AI 代理和普通 AI 应用的核心差异,再梳理基于 Python 的 Agentic 技术栈,然后给出一套从环境准备、最小代理搭建、功能测试、接口 API 化、批量任务到性能观察和问题排查的完整落地流程。适合正在做 AI 应用开发的 Python 工程师,也适合想要系统建立 Agentic AI 知识体系的同学。
1. 核心能力速览
项目标题指向的内容是“用 Python 构建真实 AI 代理的工程实践”,而不是某个具体的开源仓库或一键启动工具。所以这里的能力速览更多是从工程体系角度来评估这套内容能带给你的东西。
| 能力项 | 说明 |
|---|---|
| 项目类型 | Python 工程实践类书籍/课程,主题是 Agentic AI Engineering |
| 核心内容 | AI 代理架构、工具调用、多代理协作、记忆管理、可观测性、安全性 |
| 技术语言 | Python |
| 关键框架 | LangChain、LangGraph、CrewAI、Pydantic AI、OpenAI Function Calling 等 |
| 模型形态 | 支持云端模型 API,也支持本地模型(如 Ollama 等)接入 |
| 运行环境 | 普通开发机即可开始,本地模型场景对显存有额外要求 |
| 启动方式 | 不涉及一键包,需要按 Python 工程方式创建虚拟环境、安装依赖、运行脚本 |
| 接口 API | 工程实践通常会把代理封装为 Web API,便于外部系统调用 |
| 批量任务 | 可基于消息队列或脚本循环实现批处理 |
| 适合读者 | Python 工程师、AI 应用开发者、想系统学习 Agentic AI 的进阶学习者 |
从标题看,这本书的核心不是教你某个框架的 API 怎么背,而是帮你建立一套“怎么把一个代理从原型做成真实系统”的工程方法。真实系统的关键点包括:代理能不能稳定调用外部工具、多轮对话状态会不会丢、出错了能不能自动恢复、上线之后能不能观测和排查、以及涉及用户数据和外部系统操作时有没有安全边界。
2. Agentic AI 工程为什么值得单独学
很多人会问:大模型已经有了,LangChain 也封装得很好,为什么还要谈“工程”?
因为真实场景和 Demo 差距很大。
Demo 阶段你只需要一个 Prompt、一个模型调用、一次结果打印。但真实代理要面对的是:模型输出可能不符合 JSON 格式、工具可能调用超时、外部 API 可能返回错误、上下文可能越来越长导致成本失控、用户可能连续追问“刚才那个结果是什么意思”、代理可能在没有权限的情况下尝试执行危险操作。
这些都不是模型能力问题,而是工程问题。
Agentic AI 和普通 AI 应用的差异主要体现在四个方面。
第一,Agent 有工具使用权。普通 AI 应用只做文本生成,Agent 可以调用搜索、读写文件、执行代码、操作数据库。工具调用一旦接入,错误传播路径就变长了,任何一个环节失败,整个链路都可能中断。
第二,Agent 有状态。对话过程中的目标、上下文、中间结果、工具返回内容,都需要管理。多代理场景下,不同代理之间的状态还要同步。状态管理做不好,代理就会“失忆”。
第三,Agent 有自主决策循环。大模型不是一次性输出,而是“思考 -> 调用工具 -> 观察结果 -> 再思考”的循环。这个循环如果缺少终止条件,就会出现死循环,token 消耗会非常快。
第四,Agent 需要可观测性。你要能回答:这个代理刚才为什么调了那个工具?token 花在了哪里?哪个环节最慢?如果这些信息不可见,上线后出了问题就只能靠猜。
书名里“Engineering”这个词,强调的正是这套能力。
3. 这本书/课程能帮你建立的核心能力
从工程实践角度拆解,这类内容通常围绕下面几条能力线展开。
3.1 构建可复用的 AI 代理结构
不是每次写一个脚本把 Prompt 拼接进去,而是把代理抽象成可复用的组件:模型接口层、工具注册层、对话管理层、输出解析层。这样换模型、加工具、改 Prompt 都不会牵一发动全身。
3.2 设计与实现工具调用
工具调用是 Agentic AI 的核心能力。真实工程中要考虑:工具的描述怎么写模型才能理解、工具参数如何做类型校验、工具调用失败后如何重试、工具返回结果太长如何截断、多个工具之间如何编排。
3.3 多代理协作
复杂任务靠单个代理往往搞不定。多代理架构会让一个“主管代理”把任务拆给搜索代理、代码代理、内容审核代理,最后汇总结果。多代理的关键问题是:任务如何拆解、上下文如何传递、结果如何合并、代理之间如何避免无限互相调用。
3.4 记忆管理与上下文控制
大模型上下文窗口有限,即使支持很长的上下文也意味着更高成本。工程化的做法是引入摘要记忆、向量记忆、滑动窗口等机制,让代理只保留对当前任务重要的信息。
3.5 结构化输出与数据校验
代理的最终输出不能是一段“可能正确”的文本,而应该是可被下游系统直接消费的 JSON 或结构化数据。实践上会用 Pydantic 做输出模型定义,要求模型按指定 Schema 返回,并在解析失败时自动修正。
3.6 可观测性与调试
工程级代理必须能记录完整的运行轨迹:模型输入、模型输出、工具调用、耗时、token 消耗。更完善的方案会把轨迹可视化,方便开发者在代理“跑飞”之后回溯原因。
4. 基于 Python 的 Agentic AI 技术栈梳理
既然标题提到 Python,这里把当前常见的 Agentic AI 工程化技术栈梳理一遍。不是每个项目都要全套使用,但理解工具定位非常必要。
| 技术组件 | 定位 | 常见选项 |
|---|---|---|
| 基础模型 | 对话、推理、JSON 输出 | OpenAI、Anthropic、通义、DeepSeek、本地 Ollama 等 |
| Agent 编排框架 | 管理代理循环、工具调用、状态 | LangChain、LangGraph、CrewAI、Pydantic AI、OpenAI Agents SDK |
| 工具层 | 搜索、代码执行、文件操作、数据库 | Tavily、SerpAPI、Subprocess、SQLAlchemy、Requests |
| 记忆层 | 短期对话状态、长期知识存储 | Redis、SQLite、向量数据库(Chroma、FAISS、Milvus) |
| API 服务层 | 把代理暴露成 HTTP 接口 | FastAPI、Flask |
| 批量任务层 | 异步处理大量任务 | Celery、Redis Queue、Arq、脚本并发池 |
| 可观测性 | 追踪代理轨迹、token、耗时 | Langfuse、LangSmith、自建日志系统 |
这些组件的选择取决于你的具体场景。个人学习阶段,最简单的方式是“一个模型 API + 一类工具 + 一个编排框架 + FastAPI”,先跑通端到端流程,再逐步引入记忆、队列和可观测性组件。
5. 环境准备与前置条件
即使没有现成的一键安装包,Agentic AI 工程对开发环境的要求也比较简单。下面是一套通用准备流程,实际项目按自己的情况调整。
5.1 基础环境检查
开始之前建议确认以下几点:
- 操作系统:Windows 10/11、Ubuntu 20.04+、macOS 都可以。
- Python 版本:推荐 3.10 及以上,部分 Agent 框架对 3.11/3.12 支持更好。
- 包管理器:pip 即可,复杂项目建议使用 Poetry 或 uv。
- 模型访问:准备一个模型 API Key,或者本机通过 Ollama 启动本地模型。
- 网络连通性:确保可以访问目标模型 API 和工具 API。
python --version pip --version如果 Python 命令无法输出,需要先安装 Python 并加入系统 PATH。
5.2 创建虚拟环境
每个 Agentic AI 项目尽量使用独立虚拟环境,避免依赖冲突。
# 进入项目目录 mkdir agentic_ai_project cd agentic_ai_project # 创建虚拟环境 python -m venv .venv # 激活环境,Windows .venv\Scripts\activate # 激活环境,Linux/macOS source .venv/bin/activate激活后,命令行前缀会变成(.venv),表示已经进入虚拟环境。
5.3 安装核心依赖
根据项目技术栈安装依赖。这里以一个典型的 Agent + FastAPI 工程为例:
pip install openai langchain langgraph pydantic fastapi uvicorn requests如果你的项目计划使用本地模型,还要安装开源模型运行工具。Ollama 是比较常见的选择,安装后拉取一个模型再调用:
ollama pull qwen2.5需要注意,本地模型不能直接替换所有云端模型能力。显存占用要看你加载的具体模型参数量,7B 模型和 70B 模型差距很大,实际占用以本机nvidia-smi观察为准。
6. 从零搭建一个最小 AI 代理工程
下面写一个不依赖重型框架的最小 Agent 循环,帮助你理解 Agentic AI 的核心结构:模型调用、工具注册、代理循环。这段代码是演示性质,不是书籍附带源码,但结构上体现了工程化代理的基础。
6.1 定义工具
先定义一个计算工具和一个天气查询工具。真实项目中,工具可以是搜索、数据库查询、文件处理等任意外部能力。
import json from typing import Callable, Dict def calculate(expression: str) -> str: """计算数学表达式,比如 '1 + 2 * 3'。""" try: result = eval(expression, {"__builtins__": {}}, {}) return f"计算结果: {result}" except Exception as exc: return f"计算失败: {exc}" def get_weather(city: str) -> str: """模拟天气查询工具。""" # 真实项目应调用天气 API,这里仅演示结构 return f"{city} 的天气:晴,26 摄氏度" TOOLS: Dict[str, Callable] = { "calculate": calculate, "get_weather": get_weather, }6.2 编写代理循环
代理循环的核心逻辑是:把用户问题和工具描述一起交给模型,模型选择调用某个工具,代码执行工具后把结果返回给模型,模型基于结果生成最终答案。
from openai import OpenAI client = OpenAI() # 需要配置 API Key SYSTEM_PROMPT = """ 你是一个 AI 代理,可以调用工具解决问题。 工具列表如下: - calculate: 计算数学表达式,参数为 expression - get_weather: 查询天气,参数为 city 如果需要调用工具,请严格按照 JSON 格式返回: {"tool": "工具名", "args": {"参数名": "参数值"}} 如果不需要调用工具,直接返回答案文本。 """ def run_agent(user_input: str) -> str: messages = [{"role": "system", "content": SYSTEM_PROMPT}] for _ in range(5): # 限制循环次数,避免死循环 messages.append({"role": "user", "content": user_input}) response = client.chat.completions.create( model="gpt-4o-mini", messages=messages, ) content = response.choices[0].message.content.strip() try: action = json.loads(content) tool_name = action["tool"] tool_args = action["args"] tool_result = TOOLS[tool_name](**tool_args) messages.append({"role": "assistant", "content": content}) messages.append({ "role": "user", "content": f"工具返回结果:{tool_result}\n请根据这个结果给用户最终答案。" }) except json.JSONDecodeError: return content return "已达到最大循环次数,请简化任务或检查工具调用。" if __name__ == "__main__": user_text = input("请输入你的问题:") print(run_agent(user_text))这个示例虽然简单,但暴露了几个真实工程问题:循环次数怎么限制、工具调用异常怎么恢复、模型不按 JSON 格式输出怎么办、上下文越来越长怎么处理。这些正是 Agentic AI 工程要解决的问题。
7. 功能测试与效果验证
代理不是跑通一次就算成功,真实项目需要一套可重复的功能测试方案。
7.1 工具调用正确性测试
给代理一个明确需要调用工具的问题,检查它是否选择了正确的工具并返回正确结果。
| 测试目的 | 输入示例 | 预期结果 | 判断标准 |
|---|---|---|---|
| 工具选择 | “帮我算一下 123 * 456” | 调用 calculate | 返回计算结果 56088 |
| 工具参数 | “北京今天天气怎么样” | 调用 get_weather | 参数 city 为“北京” |
| 不调用工具 | “你好” | 直接回答 | 不进入工具调用分支 |
7.2 错误恢复测试
真实代理常常遇到工具调用失败的情况。测试时人为让工具抛异常,观察代理能否把异常信息返回给模型并继续处理。
# 模拟工具异常 def broken_tool(): raise RuntimeError("外部服务不可用")预期行为是:代理返回失败原因,并在提示中说明当前无法完成该任务,而不是直接崩溃。
7.3 多轮状态测试
连续提问两次,第二次问题依赖第一次的上下文。例如先问“北京到上海机票价格”,再问“那高铁呢”,之后检查回答是否包含对“高铁”的正确理解。
7.4 输出格式与稳定性测试
对同一问题重复运行 10 次,检查 JSON 输出是否始终可解析、字段是否完整。工程实践中常用 Pydantic 定义输出 Schema,并用 retry 机制对解析失败的结果重新生成。
7.5 成本与 token 消耗测试
记录每次运行的 token 总数、工具调用次数、请求耗时。这样可以量化每个任务的成本,为后续优化提供依据。
8. 接口 API 化与批量任务
真实系统不会只在命令行里跑代理,通常要提供 HTTP 接口,同时支持批量处理。
8.1 用 FastAPI 封装代理接口
下面把上面的run_agent包装成 POST 接口。
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class AgentRequest(BaseModel): user_input: str session_id: str = "default" class AgentResponse(BaseModel): result: str session_id: str @app.post("/api/agent", response_model=AgentResponse) async def agent_endpoint(req: AgentRequest): result = run_agent(req.user_input) return AgentResponse(result=result, session_id=req.session_id)启动服务:
uvicorn main:app --host 0.0.0.0 --port 8000接口就绪后用 curl 测试。
curl -X POST http://127.0.0.1:8000/api/agent \ -H "Content-Type: application/json" \ -d '{"user_input": "帮我算 1+1", "session_id": "test_001"}'8.2 批量任务设计
批量任务的本质是“多个请求共用同一个代理逻辑,但要隔离 session 状态”。常见做法是:读取任务文件 -> 逐条提交到代理服务 -> 收集结果 -> 写入输出文件。
import json import time import requests API_URL = "http://127.0.0.1:8000/api/agent" tasks = [ {"user_input": "计算 2 的 10 次方", "session_id": "batch_001"}, {"user_input": "上海天气如何", "session_id": "batch_002"}, {"user_input": "翻译 'hello world'", "session_id": "batch_003"}, ] results = [] for task in tasks: try: resp = requests.post(API_URL, json=task, timeout=60) results.append(resp.json()) except requests.exceptions.RequestException as exc: results.append({"task": task, "error": str(exc)}) time.sleep(0.5) # 控制请求频率 with open("results.json", "w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2) print("批量处理完成,共", len(results), "条结果")如果任务数量很大或耗时很长,需要引入异步任务队列。Celery + Redis 是常见选择:把每个任务投递到队列,后台 worker 消费任务并执行代理,前台通过任务 ID 轮询结果。这套架构要考虑失败重试、任务超时、队列积压监控,不能只写一个 for 循环。
9. 资源占用与性能观察
Agentic AI 项目的“资源”和传统 Python 服务不太一样,至少要看三层:模型侧资源、服务侧资源、token 成本。
9.1 GPU/CPU 与显存观察
如果使用云端模型 API,本地只承担逻辑编排,CPU 和内存压力不大。如果使用本地模型,显存占用会明显上升。观察显存最简单的方式:
nvidia-smiLinux 上还可以实时观察:
watch -n 1 nvidia-smi需要关注的指标有:显存使用率、GPU 利用率、显存温度。实际占用以你加载的模型参数量和推理框架为准,不要轻信别人的“固定数字”。
9.2 服务侧资源观察
代理服务本身的资源消耗主要来自并发请求数量、工具调用频率和日志写入量。CPU 高不一定代表代码有问题,也可能是大量请求同时触发了多个工具。内存持续增长通常要检查有没有把会话上下文全部留在内存里。
9.3 Token 成本观察
这是 Agentic AI 项目最容易被忽视的成本。代理循环里的每一步都在消耗 token,工具返回结果越长,下一轮请求的上下文就越大。实践中建议:
- 对工具返回内容做截断或摘要。
- 限制代理循环最大次数。
- 记录每次请求的 prompt_tokens 和 completion_tokens。
- 设置每月 token 消耗预算。
10. 常见问题与排查方法
Agentic AI 项目比普通 Web 服务更容易出现“偶发失败”,因为模型输出有随机性。下面列出一份通用排查清单。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 依赖安装失败 | Python 版本过低或包冲突 | 查看 pip 错误日志 | 升级 Python、使用虚拟环境、换用 uv 安装 |
| API Key 不可用 | Key 未配置或已过期 | 打印请求头检查 | 重新配置环境变量 |
| 模型不按 JSON 输出 | Prompt 不清晰或模型能力不足 | 打印原始输出 | 使用更强模型、增加 JSON Schema 约束 |
| 工具调用超时 | 外部 API 响应慢 | 查看工具耗时日志 | 增加超时时间、加入重试机制 |
| 代理死循环 | 缺少终止条件或工具反复返回触发信号 | 检查循环日志 | 限制最大循环次数、增加关键词终止条件 |
| 上下文超长 | 工具结果未截断、历史消息过多 | 查看请求 token 统计 | 引入摘要记忆、滑动窗口 |
| 显存不足 | 本地模型太大 | 查看 nvidia-smi | 换小模型、降低量化精度、关闭无关程序 |
| 批量任务卡住 | 队列未消费或单任务异常阻塞 | 查看 worker 日志 | 增加任务超时、失败自动重试 |
| 输出格式不稳定 | 模型随机性 | 重复运行同一请求 | 降低 temperature、增加输出 Schema 校验 |
| 端口冲突 | 8000 端口被占用 | 检查端口占用 | 换端口或结束占用进程 |
11. 最佳实践与合规边界
Agentic AI 工程化不能只关注“能不能跑通”,还要关注“能不能长期稳定运行”和“是否在安全边界内”。
11.1 工程实践建议
第一,第一次先小参数测试。不要一上来就接几十个工具、几百条任务。先把一个工具调通,再把代理循环跑通,最后再上并发。
第二,保留一套最小可运行配置。把模型 API 地址、Key、工具列表、循环次数最大限制都写进配置文件,方便团队其他成员复现。
第三,模型文件、输入素材、输出结果分目录管理。示例结构:
agentic_ai_project/ ├── configs/ # 配置文件 ├── tools/ # 工具实现 ├── agent/ # 代理逻辑 ├── api/ # FastAPI 接口 ├── data/ │ ├── inputs/ # 输入任务 │ └── outputs/ # 输出结果 └── logs/ # 运行日志第四,批量任务要加日志和失败重试。不要让一批任务的一个失败点导致整批任务白跑。
第五,接口服务要限制访问范围。代理接口通常包含模型调用权限,暴露到公网前需要加认证、限流和审计。
11.2 安全与合规边界
代理拥有工具调用能力之后,权限边界比普通应用更敏感。需要注意:
- 涉及人脸、声音、版权素材时,必须确认已获得合法授权。
- 代理如果具备操作外部系统的能力,必须有明确的授权边界,不能绕过系统权限设计。
- 用户输入和代理输出可能包含隐私信息,日志中要脱敏处理。
- 代理不应被用来规避安全限制、窃取账号、破坏系统或从事违法活动。
- 涉及商业落地时,要对代理生成的内容做人工复核,尤其是面向公众输出的场景。
12. 总结
Agentic AI Engineering 不是“给大模型套个循环”这么简单。它要求你掌握工具调用设计、状态管理、多代理协作、结构化输出、可观测性、批量任务和资源控制这套完整工程链路。
如果你准备开始,建议按这个顺序推进:先用最小代理代码跑通模型调用和工具调用,再引入 Pydantic 做结构化输出,然后把代理封装成 FastAPI 接口,最后再考虑多代理和批量任务。每加一层,都要先验证稳定性,再继续往上叠。
最容易踩的坑有两个:一是代理循环没有限制导致 token 消耗失控,二是工具调用失败后没有恢复机制导致任务直接中断。先把这两个问题解决,你的代理才具备接近“真实系统”的可靠性。
这套内容的后续扩展方向很明确:本地模型接入、向量记忆、多代理编排、代理运行轨迹可视化、以及把代理接入到自己的业务系统中。建议收藏备用,第一篇先把环境和一个最小代理跑起来。