大模型应用开发实战:从API调用到生产级AI Agent
在实际开展 AI 应用开发之前很多人的体验是“玩了 AI 才知道”学一些概念、调通一次 API就像吃了一份清淡养胃的简餐真正把大模型接进业务系统让它自主调用工具、处理上下文、稳定地对外提供接口才是“无肉不欢”的正餐。这篇文章从一个可运行的 AI Agent 示例出发带你把大模型 API 调用、工具调用、短期记忆、HTTP 服务封装、生产化改造串成一条完整链路。读完以后你可以把这个最小案例迁移到客服助手、内部知识库问答、数据查询机器人等真实场景里而不是停留在“能跑通一次 Chat Completion”。1. 先搞明白为什么“会调 API”不等于“会做 AI 应用”1.1 一个 AI 应用最少由哪几部分组成很多刚接触大模型的人第一次体验往往是这样写几行代码调用一次chat.completions.create把用户问题传进去拿到一段返回文本就认为“AI 开发”不过如此。其实这只是一次 HTTP 请求离“应用”还有很远。一个能解决实际问题的 AI 应用最少包含四层第一是模型层。你要选择用什么模型、什么参数、什么接口协议。这里决定生成质量、成本和延迟。第二是消息上下文层。模型本身是无状态的它不记得你上一次问过什么。你必须自己管理system、user、assistant、tool这几类消息并控制上下文的长度。第三是工具层。很多问题无法只靠模型“想出来”比如查询实时天气、查数据库、调用内部接口。这些能力需要通过函数调用Function Calling暴露给模型由模型决定什么时候调用、传什么参数。第四是业务接入层。你还要考虑如何把能力封装成 HTTP API、如何做权限校验、如何处理并发、如何记录日志以及模型异常时业务系统怎么兜底。如果你把“会调 API”当成“会做 AI 应用”后面遇到的每个问题都会让你觉得无从下手。真正理解这四层才算进入正题。1.2 从“清淡养胃”到“无肉不欢”的四个阶段AI 应用开发的学习路径可以分成四个阶段。第一阶段是基础调用。把一次对话跑通理解messages的结构能调整temperature、max_tokens等参数。第二阶段是结构化输出。让模型返回 JSON而不是一段自由文本这样业务代码才能稳定解析。实现方式可以是提示词约束也可以用 JSON Schema 或函数调用。第三阶段是工具调用。这是从“聊天机器人”走向“AI Agent”的关键一步。模型可以根据用户问题生成tool_calls你的程序去执行真实函数再把结果返回给模型让它生成最终回答。第四阶段是生产化。包括环境隔离、配置外置、超时重试、日志追踪、成本控制、安全过滤、监控告警。没有这一阶段项目只能跑在本地不能上线给别人用。后续章节会按这个路径展开。示例代码使用 Python 和 FastAPI接口协议采用当前最常见的大模型 Chat Completions 协议。如果你用的是 Spring AI 或其他框架核心思路一致只是 SDK 和写法不同。2. 环境准备先搭一个能复现的最小工程2.1 前置环境与依赖清单为了不把时间浪费在环境问题上建议使用 Python 3.10 或更高版本。下面这份依赖清单覆盖了 Web 服务、大模型 SDK 和配置管理依赖用途版本建议fastapi提供 HTTP 接口0.115 或更高uvicornWeb 服务进程0.34 或更高openai调用 Chat Completions 接口1.59 或更高python-dotenv读取.env配置1.0 或更高pydantic请求参数校验2.10 或更高如果直接使用最新稳定版在项目目录执行pip install fastapi uvicorn[standard] openai python-dotenv pydantic也可以把这些依赖写入requirements.txtfastapi0.115 uvicorn[standard]0.34 openai1.59 python-dotenv1.0 pydantic2.102.2 项目结构设计为了后面扩展方便不要把所有代码都写进一个文件。建议先建立这样的目录结构ai_agent_demo/ ├── app.py # FastAPI 入口 ├── agent.py # Agent 核心逻辑 ├── tools.py # 工具函数 ├── config.py # 配置读取 ├── requirements.txt ├── .env.example └── README.md采用这种结构的原因很直接config.py负责集中读取配置tools.py只放工具函数agent.py只处理模型交互app.py只负责 HTTP 层。这样以后加工具、换模型、加接口都不需要推倒重来。2.3 环境变量和配置文件API Key 不能硬编码在代码里。本地开发可以用.env文件管理生产环境应使用密钥管理服务或容器环境变量。先创建.env.example作为模板OPENAI_API_KEYyour_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-mini TEMPERATURE0.7 MAX_TOKENS1024 MAX_HISTORY_MESSAGES20然后编写config.pyimport os from dotenv import load_dotenv load_dotenv() class Settings: OPENAI_API_KEY os.getenv(OPENAI_API_KEY, ) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) LLM_MODEL os.getenv(LLM_MODEL, gpt-4o-mini) TEMPERATURE float(os.getenv(TEMPERATURE, 0.7)) MAX_TOKENS int(os.getenv(MAX_TOKENS, 1024)) MAX_HISTORY_MESSAGES int(os.getenv(MAX_HISTORY_MESSAGES, 20)) settings Settings()关键点有两个一是把模型名、温度、最大 token 数都变成可配置项二是通过load_dotenv()把.env里的内容加载到进程环境变量。实际部署时如果系统已经注入了环境变量代码不需要改动。注意OPENAI_BASE_URL可以指向兼容 OpenAI 协议的其他大模型服务商。不同服务的模型名不一定相同接入时要以服务商文档为准。完成这步后可以执行一个最简单的检查确认python -c from config import settings; print(settings.LLM_MODEL)能正常输出模型名。3. 用 Python 实现一个带工具调用的 AI Agent3.1 核心概念消息、角色和函数调用要让大模型帮你完成真实任务需要先理解消息结构。在 Chat Completions 协议中对话由一组消息组成。system用于设定人设和规则user表示用户输入assistant表示模型生成的内容tool表示工具执行后返回的结果。当你的程序告诉模型“有哪些工具可用”之后模型在回答中可能返回tool_calls。它表示我需要调用某个工具工具的入参是什么。此时你的程序不应该直接把这条消息当作最终回答而是要执行对应的工具函数再把执行结果以roletool的消息追加回去。模型看到工具结果后才会生成最终回复。这就是工具调用的基本循环。理解了这一点后面的代码就顺理成章。3.2 第一步先实现一个基础对话函数先从最简版本开始。创建一个agent.py用OpenAISDK 完成一次基础对话from openai import OpenAI from config import settings client OpenAI( api_keysettings.OPENAI_API_KEY, base_urlsettings.OPENAI_BASE_URL, ) def chat_once(user_input: str) - str: response client.chat.completions.create( modelsettings.LLM_MODEL, messages[ {role: system, content: 你是一个有用的AI助手。}, {role: user, content: user_input}, ], temperaturesettings.TEMPERATURE, max_tokenssettings.MAX_TOKENS, ) return response.choices[0].message.content这个函数虽然能跑但它没有任何记忆也无法调用外部工具。用它做简单问答可以但做不了真正有用的业务功能。3.3 第二步定义工具并执行函数调用在项目目录下新建tools.py先放两个工具函数。天气函数这里只是示例实际项目应该替换成调用真实天气服务import datetime def get_current_time() - str: return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_weather(city: str) - str: # 实际项目中这里应调用真实天气服务并对城市名做校验。 return f{city}天气晴25摄氏度示例数据然后在agent.py中定义工具描述并实现工具分发import json from config import settings from tools import get_current_time, get_weather client OpenAI( api_keysettings.OPENAI_API_KEY, base_urlsettings.OPENAI_BASE_URL, ) TOOLS [ { type: function, function: { name: get_current_time, description: 获取当前日期和时间返回字符串, parameters: { type: object, properties: {}, }, }, }, { type: function, function: { name: get_weather, description: 获取指定城市的天气, parameters: { type: object, properties: { city: {type: string, description: 城市名例如北京} }, required: [city], }, }, }, ] def call_tool(name: str, arguments: str) - dict: args json.loads(arguments) if name get_current_time: return {result: get_current_time()} if name get_weather: return {result: get_weather(args[city])} return {error: funknown tool: {name}}定义工具时parameters必须使用 JSON Schema模型会根据这段描述决定调用哪个工具、传入什么参数。工具描述写得越清楚模型调用的准确率越高。3.4 第三步实现 Agent 主循环并加入短期记忆现在实现完整的run_agent函数。它需要维护一个session_messages列表并在必要时追加 assistant 消息和 tool 结果MAX_TOOL_ROUNDS 5 def run_agent(session_messages: list, user_input: str) - str: session_messages.append({role: user, content: user_input}) for _ in range(MAX_TOOL_ROUNDS): response client.chat.completions.create( modelsettings.LLM_MODEL, messagessession_messages, toolsTOOLS, temperaturesettings.TEMPERATURE, max_tokenssettings.MAX_TOKENS, ) msg response.choices[0].message # 模型要求调用工具 if msg.tool_calls: assistant_msg {role: assistant, content: msg.content} assistant_msg[tool_calls] [ { id: tc.id, type: tc.type, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in msg.tool_calls ] session_messages.append(assistant_msg) for tool_call in msg.tool_calls: tool_result call_tool( tool_call.function.name, tool_call.function.arguments, ) session_messages.append( { role: tool, tool_call_id: tool_call.id, content: json.dumps(tool_result, ensure_asciiFalse), } ) continue # 模型没有要求调用工具说明已经生成了最终回答 assistant_reply msg.content or 没有生成文本内容。 session_messages.append({role: assistant, content: assistant_reply}) return assistant_reply return 模型多次尝试后仍未得出结果请检查工具定义或上下文。这段循环很关键。如果msg.tool_calls不为空说明模型还在“思考”你需要执行工具并把结果回传然后再次请求模型。如果直接返回给用户用户会看到一段奇怪的中间过程而不是最终答案。为了管理会话再写一个SessionStore类。它负责创建新会话、控制上下文长度from config import settings class SessionStore: def __init__(self): self._sessions {} def _init_session(self, session_id: str): if session_id not in self._sessions: self._sessions[session_id] [ {role: system, content: 你是一个有用的AI助手。} ] def get_session(self, session_id: str): self._init_session(session_id) return self._sessions[session_id] def trim(self, session_id: str): self._init_session(session_id) messages self._sessions[session_id] system [m for m in messages if m[role] system] others [m for m in messages if m[role] ! system] if len(others) settings.MAX_HISTORY_MESSAGES: others others[-settings.MAX_HISTORY_MESSAGES:] self._sessions[session_id] system others为什么要限制历史消息长度大模型输入有 token 上限对话越长消耗越大延迟越高。实际项目不能无限保留历史常见的做法是只保留最近的 N 条消息或者把早期对话压缩成摘要后再继续。注意tool消息也是上下文的一部分计算历史长度时不能只数user和assistant否则上下文可能意外超长。4. 把 Agent 封装成 HTTP 服务运行验证与排查4.1 用 FastAPI 暴露/chat接口Agent 逻辑已经完成接下来把它封装成 HTTP 接口。这样前端、测试脚本或其他服务都可以接入。创建app.pyimport logging from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent import run_agent, SessionStore from config import settings app FastAPI() store SessionStore() logger logging.getLogger(uvicorn.error) class ChatRequest(BaseModel): session_id: str message: str class ChatResponse(BaseModel): reply: str app.post(/chat, response_modelChatResponse) def chat(req: ChatRequest): store.trim(req.session_id) session store.get_session(req.session_id) try: reply run_agent(session, req.message) except Exception: logger.exception(chat handler failed, session_id%s, req.session_id) raise HTTPException(status_code502, detail模型服务调用失败) return ChatResponse(replyreply)session_id由调用方传入。同一个session_id会复用一套上下文不同session_id之间互不干扰。生产环境中内存字典通常要替换成 Redis 等外部存储否则服务重启或水平扩容后会话会丢失。4.2 启动服务并验证功能安装依赖后先创建一个本地.env文件填入你的 API Key 和模型名然后启动服务uvicorn app:app --reload --port 8000启动后用curl测试基础对话curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {session_id: demo, message: 现在几点了}预期会返回类似这样的 JSON其中时间部分由模型从工具结果生成{ reply: 当前时间是 2025-01-15 14:30:00示例 }继续测试工具调用curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {session_id: demo, message: 北京天气怎么样}预期返回{ reply: 北京天气晴25摄氏度示例数据 }最后测试短期记忆。在同一个session_id下继续提问curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {session_id: demo, message: 我刚才问的是哪个城市}如果模型能答出“北京”说明历史消息正常保留Agent 已经具备基础记忆能力。4.3 常见问题与排查链路本地运行最常见的几类问题可以按下面的表格排查问题现象常见原因检查方式处理建议返回 401 认证失败API Key 错误、过期或 base_url 配置不对检查.env和日志中的错误码确认 Key 和接口地址重新生成 Key 后重试请求一直超时网络不稳定、模型服务负载高查看请求耗时和错误堆栈设置超时时间加入重试和降级逻辑上下文超限历史消息过长token 超过模型上限查看错误日志中的 token 数量减少MAX_HISTORY_MESSAGES或改用摘要压缩工具结果未被模型理解tool_call_id不匹配或 tool 消息格式错误打印完整的messages列表确认roletool的消息使用正确tool_call_id中文字符变成\uXXXX或乱码序列化时未指定 UTF-8检查 JSON 输出和日志使用ensure_asciiFalse统一文件编码为 UTF-8模型反复调用同一个工具工具返回结果不明确或描述有歧义查看连续请求日志优化工具描述限制最大调用轮数排查时建议遵循固定顺序先看输入参数是否正确再看配置和 Key 是否有效然后看请求日志和模型返回最后检查工具函数本身的异常。不要一上来就怀疑框架。4.4 日志该怎么打生产环境排查不能靠“猜”。每个接口请求最好都关联一个request_id日志中至少记录以下内容会话 ID用户问题注意脱敏模型名称是否触发了工具调用调用了哪个工具本次请求耗时返回的 token 使用量例如{ timestamp: 2025-01-15T14:30:00.000Z, request_id: req_123, session_id: demo, model: gpt-4o-mini, tool_called: get_weather, latency_ms: 850, prompt_tokens: 320, completion_tokens: 45 }有了这些信息才能回答“为什么慢”“为什么回答不对”“为什么成本涨了”这类问题。5. 生产环境还需要补哪些“肉”5.1 模型选型与参数调优学习环境可以直接用默认模型、默认参数。生产环境要根据业务场景重新选型和调参。参数作用学习环境建议生产环境建议temperature控制随机性值越大回答越发散0.7客服场景 0.2创意写作 0.8max_tokens限制单次输出最大 token 数512根据输出长度预留 20% 余量MAX_HISTORY_MESSAGES控制上下文长度20动态策略摘要压缩或向量记忆超时时间等待模型返回的最长时间30 秒5 到 10 秒配合重试重试次数面对瞬时错误的容忍度02 到 3 次并使用指数退避参数不是越大越好。temperature1.0会让回答更有创意也会让格式更容易漂移。如果业务要求模型稳定输出 JSON温度通常不要超过 0.3。5.2 流式输出、超时与降级对话类产品通常需要流式输出让用户看到“打字机”效果避免长时间空白等待。在 OpenAI SDK 中只需要把streamTrue加进去然后遍历流式事件。FastAPI 端可以使用StreamingResponse向前端推送。流式接口还需要额外处理连接断开、客户端取消、模型中断等异常不能简单套用同步返回逻辑。生产环境还要设计降级方案。比如主模型超时后可以切换到备用模型模型服务整体不可用时可以返回预设话术而不是让请求一直卡住。这个兜底逻辑一定要提前测试不能用的时候才想起来。5.3 安全、合规与数据保护这部分不能省略。即使是最简单的 AI 应用也要关注以下几点首先API Key 绝不能出现在前端代码、仓库和日志中。本地开发用.env生产环境使用密钥管理服务或容器平台的环境变量注入。其次用户输入不能直接拼进系统提示词或工具参数里。工具函数要对参数做校验例如查询天气时必须校验城市名是否在允许列表内。不要信任模型生成的参数更不要直接执行模型生成的代码。然后日志和审计要做脱敏。用户名、手机号、身份证号、内部系统地址等敏感信息在写入日志前要先做脱敏或直接不记录。最后要考虑提示词注入。用户可能在上一条消息里要求系统“忽略之前所有指令”。如果模型有执行工具的能力系统提示词里要明确约束“只能按业务规则调用工具”并且在工具层做权限校验。5.4 可观测性与版本回滚上线前要确认以下问题都有人盯着模型调用的成功率是多少平均延迟和 P95 延迟是多少单次会话平均消耗多少 token哪些会话触发了工具调用有没有成本突增的 session模型返回格式错误率是多少推荐在接口层统一记录指标并接入 Prometheus、Grafana 或云厂商的监控体系。模型调用失败时要能快速切换模型或回滚到旧版本。因此模型名称、提示词版本、工具函数版本都要纳入发布管理最好和代码一起走 CI/CD而不是直接改线上配置。6. 从示例项目延伸AI 开发最佳实践清单6.1 后续可以扩展的方向这个最小 Agent 已经具备三个核心能力对话、工具调用、短期记忆。继续往下走常见方向有把内存SessionStore替换为 Redis支持多实例部署。加入 RAG把公司文档切分、向量化后放到向量数据库让模型基于文档回答。引入多工具编排让 Agent 可以查询订单、创建工单、发送通知。如果团队以 Java 为主可以研究 Spring AI它提供了类似ChatClient和函数调用的抽象思路和本文一致。对模型返回做结构化校验强制输出 JSON Schema减少下游解析故障。建议不要一上来就搭建非常复杂的 Agent 框架。先把手写工具调用的循环跑明白再去用框架你会更清楚框架到底帮你解决了什么问题。6.2 发布前检查清单无论项目大小上线前都可以对照这份清单过一遍[ ] API Key 是否通过环境变量或密钥管理注入是否有轮换流程。[ ] 模型名称是否正确是否区分了开发环境和生产环境。[ ] 超时时间、重试次数是否配置合理是否做了降级。[ ] 上下文长度是否受限超长时会不会报错。[ ] 工具函数是否有参数校验、异常处理和超时控制。[ ] 是否记录 request_id、模型、token 消耗和延迟。[ ] 日志是否脱敏用户敏感数据是否被写入日志。[ ] 是否有输入输出过滤提示词注入是否有缓解方案。[ ] 是否设置成本告警单会话 token 是否有限额。[ ] 模型返回异常时用户看到的是不是友好提示。6.3 一个值得长期坚持的练习方法建议从复制这篇文章的代码开始然后做三个改动加一个自己的工具函数比如“查询本周待办”把历史会话存储从内存换成 Redis把/chat接口改成流式返回。这三个改动做完你会真正理解模型、工具和业务三者之间的边界。到这一步你就不再是“会调 API”的旁观者而是能独立搭建 AI 应用的开发者。之后再去看 LangChain、Spring AI 的源码或文档会发现很多概念似曾相识学习速度会明显变快。AI 应用开发和其他后端开发一样难点不在大模型本身而在如何把模型干净、稳定、可控地嵌进现有系统。清淡养胃的部分是概念和 API无肉不欢的部分是工程实现。两者缺一不可但从工程实践入手往往更能建立长期能力。
