尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

Python构建真实AI代理:Agentic AI工程实践全解析

Python构建真实AI代理:Agentic AI工程实践全解析 这次我们来看一个很典型的工程向主题使用 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、MilvusAPI 服务层把代理暴露成 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( modelgpt-4o-mini, messagesmessages, ) 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_modelAgentResponse) async def agent_endpoint(req: AgentRequest): result run_agent(req.user_input) return AgentResponse(resultresult, session_idreq.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: 帮我算 11, 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, jsontask, timeout60) 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, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) 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 消耗失控二是工具调用失败后没有恢复机制导致任务直接中断。先把这两个问题解决你的代理才具备接近“真实系统”的可靠性。这套内容的后续扩展方向很明确本地模型接入、向量记忆、多代理编排、代理运行轨迹可视化、以及把代理接入到自己的业务系统中。建议收藏备用第一篇先把环境和一个最小代理跑起来。
返回列表