这次我们来看一个偏工程落地的 Python 方向Agentic AI Engineering也就是用 Python 从零搭一个具备真实任务执行能力的 AI 代理AI Agent。网上讲 Agent 概念的文章很多但真正能把“模型调用、工具注册、任务规划、状态管理、接口暴露、批量执行”串成一套完整工程的项目并不多。这个主题的价值就在这里它不是教你调一个 Prompt而是把 AI 代理当成一个正经软件系统来设计和实现。先给结论如果你已经会 Python 基础语法想在本地把 AI 代理跑起来并接入自己的数据源、工具链和业务接口这个方向非常适合。它会覆盖从 LLM 调用、工具函数注册、ReAct 循环、记忆管理到 FastAPI 接口封装的全流程。接下来我按工程实践的顺序把环境准备、核心架构、代码实现、接口调用、批量任务、性能观察和排查方法完整拆开讲一遍。1. 核心能力速览在动手之前先把这个方向的能力边界和门槛列清楚。下面这张表基于常见的 Agentic AI 工程实践整理具体参数要以你实际安装的库版本和模型推理方式为准。能力项说明核心语言Python 3.10建议 3.11 或 3.12主要依赖LangChain / LangGraph、OpenAI SDK、Pydantic、FastAPI、uvicorn模型接入OpenAI 兼容接口、本地 Ollama、vLLM、DeepSeek、Qwen、Llama 等显存需求纯 API 调用无显存要求本地模型需按模型参数量评估7B 量化模型一般 6G~8G 显存起步启动方式命令行脚本 FastAPI 服务核心功能工具调用、任务规划、多步推理、记忆管理、批量执行、API 接口是否支持 API支持可封装为 HTTP 服务是否支持批量任务支持可设计队列逐条处理适合场景本地自动化、知识库问答、数据分析助手、定时任务、个人 AI 工作流门槛评估代码能力要求中等需要理解函数装饰器、异步编程和 JSON Schema从材料看这个主题的关键词集中在 Python、Agentic AI、AI 代理、AI 工程说明核心不是某个单一模型而是一套工程方法。实际学习时最值得投入精力的是三个点工具注册机制、Agent 循环控制和状态持久化。2. 适用场景与使用边界AI 代理听起来很通用但工程上用起来有明确的边界。适合做的场景包括把多个内部工具串成一个工作流比如查数据库、调接口、发通知给知识库问答增加“能动手”的能力比如查完资料后生成汇总报表替代重复的脚本调度让模型根据你的自然语言描述决定调用哪个函数以及作为个人助手把日常的文本处理、文件整理、信息检索任务自动化起来。不适合的场景也要说清楚。涉及太高实时性和安全性的操作比如直接执行生产环境 SQL、操作系统级文件删除、资金交易不建议交给完全自主的代理去执行必须加人工确认节点。另外代理对上下文长度敏感长任务容易丢失中间状态如果任务链条超过十几步需要考虑状态保存和断点恢复。合规边界是重点。如果你要接入企业数据、用户隐私信息或版权材料必须确认授权。模型的输入输出不要包含敏感凭证工具函数内部要做好参数校验。代理调用外部 API 时需要设置超时和频控避免批量任务把目标服务打挂。这些不是可选项是工程上线的基本要求。3. 环境准备与前置条件建议在 Linux 服务器或 macOS 上开发Windows 也能跑但要注意路径和依赖兼容性。Python 版本优先选 3.11 或 3.123.10 也能用但 Pydantic v2 在 3.12 下表现更稳。3.1 创建虚拟环境先用 conda 或 python -m venv 建一个干净环境避免和系统 Python 冲突python -m venv .venv source .venv/bin/activateWindows PowerShell 下请用.venv\Scripts\Activate.ps13.2 安装核心依赖pip install --upgrade pip pip install langchain langgraph openai pydantic fastapi uvicorn python-dotenv requests如果你打算接本地模型还需要装 ollama 或 vllm 对应的客户端库。这里我给的是通用模板实际版本请以官方发布为准。安装完成后建议先跑一句检查命令python -c import langchain, langgraph, fastapi; print(deps ok)能看到 deps ok 说明基础环境没问题。3.3 模型接入准备最简单的方式是配置环境变量文件.envOPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 MODEL_NAMEgpt-4o-mini如果使用本地 Ollama 服务BASE_URL 改成http://127.0.0.1:11434/v1MODEL_NAME 改成你拉取的模型名就好。用 OpenAI 兼容协议的好处是同一个代码可以无缝切换云端模型和本地模型。4. Agentic AI 核心架构怎么设计看完环境准备接下来是工程的核心一个 AI 代理到底由哪几层组成。从设计层面拆解可以分成五个模块。第一层是模型服务层。代理本身不做决策决策依赖 LLM。这一层负责接收 Prompt 和消息历史返回结构化的模型输出。建议统一走 OpenAI 兼容协议方便切换模型。第二层是工具注册层。代理能执行什么动作取决于你给它注册了哪些函数。比如查询天气、调用搜索、读写文件、执行 Python 代码块每个工具都要有名称、描述、输入参数 JSON Schema。第三层是代理循环层。也就是 ReAct 模式模型思考下一步应该调用哪个工具代码层执行工具函数把结果返回给模型模型再决定是继续调用还是给出最终答案。这个循环是代理智能的核心。第四层是记忆层。短期记忆保存当前对话上下文长期记忆保存之前的任务结果和用户偏好。LangGraph 里的 State 机制和 Checkpointer 就是干这件事的。第五层是服务暴露层。代理跑在后台循环里没有意义要能提供给外部调用所以用 FastAPI 包一层 HTTP 接口支持同步请求和异步任务提交。这些模块不需要一开始全部做完可以先从一个最小循环开始再逐步加上记忆和服务层。5. 用 Python 实现一个最小 Agent理论讲完了直接进入代码。下面这个示例实现了一个带工具调用能力的 Agent 核心循环使用 LangChain 的工具装饰器和 LangGraph 的状态图。import json from typing import Annotated, TypedDict from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langgraph.graph.message import add_messages # 1. 定义状态 class AgentState(TypedDict): messages: Annotated[list, add_messages] # 2. 注册工具 tool def get_weather(city: str) - str: 查询指定城市的天气情况。 # 生产环境可替换成真实天气 API return f{city} 当前天气晴朗25 摄氏度。 tool def calculate(expression: str) - str: 计算数学表达式例如 12*3。 try: result eval(expression) return str(result) except Exception as e: return f计算失败: {e} tools [get_weather, calculate] # 3. 初始化模型 model ChatOpenAI(modelgpt-4o-mini, temperature0) model_with_tools model.bind_tools(tools) # 4. 构建节点 def call_model(state: AgentState): response model_with_tools.invoke(state[messages]) return {messages: [response]} # 5. 构建状态图 graph StateGraph(AgentState) graph.add_node(agent, call_model) graph.add_node(tools, ToolNode(tools)) graph.set_entry_point(agent) def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return tools return END graph.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) graph.add_edge(tools, agent) app graph.compile()运行这个 Agent 只需要传入一条用户消息def run_agent(query: str): events app.stream( {messages: [HumanMessage(contentquery)]}, config{configurable: {thread_id: test-1}} ) for event in events: for value in event.values(): last_msg value[messages][-1] if isinstance(last_msg, AIMessage) and last_msg.content: print(AI:, last_msg.content) if __name__ __main__: run_agent(北京天气怎么样顺便算一下 12*84)这段代码核心要理解bind_tools和should_continue。bind_tools把工具函数的结构化描述传给模型模型如果决定调用工具就会在返回内容里包含tool_calls。should_continue根据是否有工具调用来决定走工具节点还是直接结束。这个循环就是 ReAct 模式的最小实现。如果你不想用 LangGraph也可以手写一个 while 循环实现同样的效果代码会更啰嗦但更清晰。实际工程里用 LangGraph 的好处是状态管理、断点恢复和并行分支都有现成机制。6. 工具注册与参数校验工具层的可靠性决定了代理的可用性。在实际项目中模型可能会生成不合法参数也可能不按你的意图调用函数所以工具层必须做三件事。第一参数必须有 JSON Schema 约束。用 Pydantic 定义工具输入结构让模型严格按照结构输出参数from pydantic import BaseModel, Field class WeatherInput(BaseModel): city: str Field(description城市名称例如 北京、上海) unit: str Field(defaultcelsius, description温度单位celsius 或 fahrenheit)然后在tool装饰器上使用这个模型from langchain_core.tools import StructuredTool def get_weather_impl(city: str, unit: str celsius) - str: return f{city} 天气信息单位 {unit}当前温度为 25 度。 weather_tool StructuredTool.from_function( funcget_weather_impl, nameget_weather, description查询城市的天气温度, args_schemaWeatherInput )第二工具内部必须处理异常。任何未被捕获的异常都会直接中断 Agent 循环所以工具函数建议统一使用 try-except 把错误转成用户可以读懂的字符串返回给模型让模型决定下一步。第三工具要有返回长度上限。有些工具可能返回很大的数据比如数据库查询结果。如果全塞回给模型会快速消耗上下文窗口导致后续推理质量下降。建议在工具层做截断或摘要def truncate(text: str, limit: int 800): if len(text) limit: return text[:limit] ... (已截断) return text这三点做好之后Agent 的工具调用才真正具备工程可用性。7. 接入本地模型与复用现有 Python 工作流云端模型 API 调用快但生产环境经常要考虑数据安全和成本。你可以用 OpenAI 兼容协议切换到本地推理服务。以 Ollama 为例from langchain_openai import ChatOpenAI local_model ChatOpenAI( modelqwen2.5:7b, base_urlhttp://127.0.0.1:11434/v1, api_keyollama, temperature0 )这里的关键是base_url指向本地推理服务的 OpenAI 兼容端点api_key随便填因为本地服务不校验身份。如果你已经有现成的 Python 函数比如定时爬虫、数据清洗、报表生成要把它们接成 Agent 工具也很简单。只需要包装一层函数描述和参数 Schema让模型知道“什么时候调用、传什么参数”。这意味着你之前写的 Python 脚本不需要重写加一个装饰器就能变成代理可调用的工具能力。实际项目里我建议从这三类工作流开始接入数据处理类读取 CSV、筛选数据、计算统计值、生成图表。信息检索类调用搜索 API、读取本地文件、查询 SQLite 数据库。写入操作类发送邮件、创建文档、更新 Notion、调用公司内部接口。写入类操作务必增加一个 confirm 参数模型调用时默认 False由调用方决定是否真正执行。这能避免代理在不该执行的时候执行了破坏性操作。8. FastAPI 封装 Agent 接口Agent 循环跑通之后下一步就是把它变成服务。用 FastAPI 封装一组 HTTP 接口这样其他应用、前端或是定时任务都可以调用。8.1 基础接口代码from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uuid app FastAPI(titleAI Agent Service) class ChatRequest(BaseModel): query: str thread_id: str None class ChatResponse(BaseModel): thread_id: str answer: str app.post(/v1/chat, response_modelChatResponse) async def chat(req: ChatRequest): try: thread_id req.thread_id or uuid.uuid4().hex answer run_agent(req.query) return ChatResponse(thread_idthread_id, answeranswer) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)8.2 启动服务uvicorn main:app --host 0.0.0.0 --port 8000启动后访问http://127.0.0.1:8000/docs可以看到 Swagger 文档页面。这个页面可以直接用来测试接口。8.3 curl 调用测试curl -X POST http://127.0.0.1:8000/v1/chat \ -H Content-Type: application/json \ -d {query: 北京天气怎么样} \ --max-time 120这里的字段只是一个通用示例。实际项目里你可能会加更多控制参数比如temperature、max_steps、tools_enabled。建议把这些参数全部放进请求体让调用方能够按需调整。9. 批量任务处理与状态管理单个接口能跑通之后批量任务就是顺势要做的事。核心需求是给一批输入让代理逐条处理并且能处理失败重试。最简单的做法是写一个批量脚本循环读取输入文件逐条调用 Agent 逻辑把结果写回输出文件import json, time from concurrent.futures import ThreadPoolExecutor, as_completed def process_item(item): query item[query] try: answer run_agent(query) return {query: query, status: success, answer: answer} except Exception as e: return {query: query, status: failed, error: str(e)} with open(batch_input.jsonl, r, encodingutf-8) as f: items [json.loads(line) for line in f if line.strip()] results [] with ThreadPoolExecutor(max_workers4) as executor: future_map {executor.submit(process_item, item): item for item in items} for future in as_completed(future_map): result future.result() results.append(result) time.sleep(0.5) # 控制并发频次 with open(batch_output.jsonl, w, encodingutf-8) as f: for r in results: f.write(json.dumps(r, ensure_asciiFalse) \n) print(f完成 {len(results)} 条成功 {sum(1 for r in results if r[status]success)})批量处理有四个工程要点必须记录每个任务的输入、输出、状态和错误信息方便查日志。ThreadPoolExecutor 的并发数不要盲目调大模型 API 和本地推理服务都有吞吐上限。每条任务之间做短暂 sleep避免触发限流。对失败任务单独保存后续重跑时只处理失败项。如果你用的是 LangGraph还可以利用它的 Checkpointer 机制在任务中途保存状态。这样即使进程崩溃也可以从断点恢复不需要从头开始执行。10. 资源占用与性能观察资源占用主要看模型部署方式。使用云端 API 时本地资源占用集中在 Python 进程、工具库加载和服务端内存上显存要求为零。使用本地模型时GPU 显存占用与模型参数规模和量化精度直接相关。7B 模型 4bit 量化一般需要 6G 到 8G 显存13B 模型则需要更多空间。这里不写死以你实际运行 nvidia-smi 看到的数值为准。启动服务后开三个终端分别观察nvidia-smi -l 1每 1 秒刷新一次 GPU 显存和利用率。top观察 CPU 和内存。FastAPI 日志终端观察请求耗时和错误。影响性能的主要因素按影响程度排序为模型参数量、上下文长度、工具返回内容大小、并发请求数、工具调用次数。一个 Agent 任务如果连续调用了 5 次工具每次工具返回 2000 字那模型实际处理的 token 量会远超用户输入的文本长度。这也是为什么工具返回内容要截断。降低资源占用的经验方法# 本地模型优先选择量化后的 GGUF 格式例如 q4_k_m 版本减少不必要的历史消息发送超出窗口的消息及时裁剪或总结。把大文件读取放在工具函数内部只把处理结果返回给模型而不是把整个文件内容塞进上下文。批量任务使用固定并发数观察 API 响应时间如果响应变慢就减少并发。11. 常见问题与排查方法下面是实际开发中容易踩的坑。我把问题现象、可能原因和解决办法整理成了表格。问题现象可能原因排查方式解决方案模型不调用工具直接给答案工具描述不够明确或模型本身工具调用能力弱打印模型原始返回看是否包含 tool_calls 字段优化工具 description改用支持工具调用的模型工具参数格式错误没有为工具定义 JSON Schema检查工具函数的 docstring 和 args_schema使用 Pydantic 定义参数模型启动后 8000 端口被占用其他服务占用了端口lsof -i :8000或netstat -ano更换端口启动本地 Ollama 模型响应慢显存不足或模型被换出nvidia-smi 观察显存换更小模型或量化版本批量任务中途卡住网络超时或模型 API 限流查看日志最后一条输出设置 requests 超时增加重试逻辑LangGraph 状态错乱没有使用 checkpointer 管理 thread_id检查 config 中的 thread_id每次对话使用独立 thread_id中文输出乱码终端编码或 API 编码设置问题检查返回字符串统一使用 utf-8 编码避免 print 特殊字符Python 环境本身也常出问题。常见的pip install慢或失败时可以切换国内镜像源pip install -i https://pypi.tuna.tsinghua.edu.cn/simple langchain langgraph如果依赖版本冲突先看报错信息中提示的包名尝试升级或降级对应包不要盲目卸载重装。12. 最佳实践与工程建议最后给一组工程建议这些是代理从“能跑”走向“能用”的关键细节。12.1 先做最小闭环第一次不要急着做复杂的多功能代理。先跑通一个只有两个工具的 Agent 循环确认模型能正确调用工具再逐步加节点。每增加一个工具都要单独测试该工具的参数传递和异常处理。12.2 保持一个可复现的配置模板把模型名、API 地址、温度参数、超时时间、并发数全部放进.env或配置 JSON 文件不要写死在代码里。有利于切换模型和部署环境。12.3 设计清晰的工具边界工具尽量只做一件事不要设计一个“全流程工具”否则模型难以判断何时调用。工具描述要写清楚执行条件和副作用。12.4 增加日志和追踪建议记录每次代理运行的完整轨迹用户输入、工具调用、参数、工具返回、模型最终输出。这样出了问题可以完整复盘。12.5 人工确认写入操作凡是涉及发送消息、修改数据、删除文件的操作都建议加一个确认机制。要么在工具内部要求confirmTrue参数要么在业务层做二次确认。12.6 合规与安全接入真实业务数据前确认数据使用范围和授权。不把 API Key 写进代码仓库使用环境变量管理敏感信息。批量调用外部服务时控制频率和并发避免给对方造成压力。涉及人脸、声音、版权素材等信息时必须确认授权并明确结果的使用边界。13. 总结与下一步Agentic AI 这个方向最值得投入时间去掌握的是 ReAct 循环、工具注册机制、状态管理和服务封装这四块掌握之后你就能把任意的 Python 脚本改编成 AI 代理能力。建议拿到环境后先验证三件事第一模型能否稳定调用你注册的两个工具第二工具返回错误时代理能不能恢复正常回答第三FastAPI 接口能否被外部程序稳定调用。这三关过了这个代理就具备实际使用的骨架了。最容易踩的坑是模型上下文被工具返回内容塞满导致多轮任务后效果变差。所以工具返回内容要尽早做截断这是多数初学者忽略的点。接下来可以扩展的方向有很多给代理加记忆数据库、接入本地知识库做 RAG、用 LangGraph 画更复杂的并行分支、把代理部署成 Docker 服务、加一个前端对话界面。每一步都能让代理从“玩具”变得更接近产品级应用。建议从“给代理加一个 SQLite 长期记忆”开始这个改动会直接提升代理的实用价值。