AI工程化落地指南：从Prompt设计到Agent服务化-育师

从零做AI工程，最容易被误解的一件事是：以为工作的重心是训练模型。实际上，绝大多数项目并不需要从权重开始写起，而是要把现成的大模型能力稳定地接进业务流程里。这个“接”的过程，就是AI工程的日常。我亲眼看过很多同学把模型API跑通就宣布“交付”，结果用户一上量，超时、乱答、token爆炸、结果不可复现，全崩在线上。今天这篇文章，我就结合从零起步做AI项目的真实经历，把一个完整的AI工程化链路拆开讲清楚：从技术选型、Prompt设计、函数调用，到Agent主流程、缓存、服务化，再到多模型协作和效果评估。内容偏实战，代码可直接参考，适合刚接触AI工程、想系统搭建AI应用的算法工程师、后端开发者和独立开发者。

1. 整体设计：AI工程化到底在解决什么问题

1.1 从Demo到系统的巨大落差

“AI工程从零开始”听起来像是一个起点很低的事，但真正的难点从来不是把模型接口翻出来调通。Demo阶段你只需要一个Prompt、一条消息、一个答复。生产环境需要面对的是：输入内容千变万化、调用结果不稳定、业务逻辑需要和模型决策穿插执行、同一问题重复问会反复消耗token，以及没有一套度量方式能告诉你好还是不好。

我从一个工单分类需求开始复盘。最初的需求非常朴素：把用户提交的工单自动分类，再生成一段回复建议。看起来就是一个“文本分类+生成”的组合。但一旦落地，马上会遇到如下问题：工单文本可能包含错别字、口语化表达、多问题混合；分类必须和业务线严格对应，不能自由发挥；回复建议需要调用用户的历史订单数据，而不是凭空捏造；每一次分析都要有日志可查、有成本可估。

所以工程化的第一步不是选模型，而是把用户的需求拆成“数据流”和“状态流”。数据流负责把工单文本、用户信息、业务规则传给模型；状态流负责记录模型已经做了哪些判断、还需要调用哪些工具去补哪些信息。缺了系统性的切分，后面无论用多强的模型都会失控。

1.2 先定约束，再选模型服务

我一般会先列几个约束条件，再选模型服务，而不是反过来追热门大模型。需要评估的维度包括：上下文长度、推理质量、接口稳定性、并发上限、计费模式、数据合规。对于多数业务系统，模型能力只需要“够用”，但稳定性必须“顶得住”。

下面是我常用的选型参考维度：

维度	说明	落地建议
上下文长度	一次请求能容纳多少文字	至少8K起步，4K经常不够用
推理质量	对指令理解和格式遵循能力	用业务内的10个真实案例做盲测
接口稳定性	可用性、限流、响应速度	重点看QPS上限和失败率
计费方式	按token还是按次	按token更适合问答，按次适合固定流水线
数据合规	数据是否出域、是否可追溯	本地私有化部署往往更稳

我目前常用的是国内几家主流的大模型API服务，例如阿里云百炼、智谱AI、DeepSeek等。它们的优势是文档齐全、兼容OpenAI格式、生态工具链多，接入成本低。对于大多数AI工程来说，真正掌握“如何把模型用对”，远比“用哪个模型”重要。模型会迭代，而工程经验是复用的。

1.3 最小架构应该怎么切

一个可以长期维护的AI应用，至少要划分成五层：接入层、模型层、记忆层、工具层、评估层。

接入层负责统一接收请求、鉴权、限流、记录日志。模型层负责和具体的模型服务交互，包括重试、超时、降级。记忆层负责保存对话历史、业务上下文、短期缓存和长期知识库。工具层负责把模型需要执行的函数、API、数据库查询统一注册成可调用工具。评估层则在旁路记录每一次模型输出，跑规则校验、人工打分或模型打分。

这个分法不是拍脑袋，它决定了后续出问题时你能在哪里加日志、在哪里换模型、在哪里加缓存。如果所有逻辑都堆在一个文件里，表面上进度快，一旦请求量上来，连“这条结果是谁生成的”都查不清。

2. 核心参数与Prompt工程实操

2.1 模型参数不是玄学

大模型服务通常会开放温度、Top P、Max Tokens、停止符等参数。很多新手上来就是“照着别人的配置填”，不理解每个参数背后的意义，结果输出要么太散，要么总被截断。

温度（temperature）控制随机性。取值0到2之间，越高越天马行空，越低越稳定。做分类、抽取、格式化输出，我一般调到0.1到0.3。做创意文案，会在0.7到0.9之间。你要是做代码生成或者数据清洗，建议直接往0靠近。

Top P控制候选词的概率累计。它和温度是配合关系，不是叠加关系。我在实践中习惯只用温度做主要控制，把Top P固定在0.8左右，避免两个随机参数互相打架，导致输出诡异。

Max Tokens很多人当成“回答长度”，其实它的作用是“截断上限”。它会限制生成的最大token数量。如果任务需要模型对长文本做总结，而Max Tokens设得比原文还短，结果必然被截断，甚至产生半截JSON。我建议把Max Tokens设置为预估值的一点五倍以上。

停止符（Stop）是个很实用的参数。你可以在生成遇到特定字符串时停止，比如JSON解析场景可以设置“```”作为停止符号，避免代码块包裹干扰解析。但请注意，停止符不是每个模型服务都支持，使用前先读文档。

2.2 Prompt模板的结构化写法

Prompt Engineering不是把需求用大白话塞给模型，而是让模型的输出空间被约束到你的业务空间内。我总结了一个简单的模板结构：角色定义 + 任务目标 + 输入信息 + 输出约束 + 少数示例。

举个工单分类的例子，下面是一段有效的Prompt模板：

你是一名客服工单处理助手。请根据用户的工单描述，完成两项工作： 1. 将工单分类到以下类别之一：退款、物流、质量、安装、售后、咨询。 2. 基于用户历史订单信息，生成一段不超过50字的回复。 工单描述：{ticket_text} 用户历史订单信息：{order_info} 请严格以JSON格式输出，格式如下： {"category": "类别", "reply": "回复内容"}

这里的关键是“让模型没有自由发挥的余地”。每个变量都给清楚，每个输出字段都定义好。少了示例的时候，模型偶尔会给出完全合理但不符合你格式要求的结果。一旦加上一个示例，格式正确率会明显提升。所以模板里至少要有“一正一反”两个示例，尤其要把反例讲明白。

2.3 用函数调用让模型可控制

光会写Prompt还不够。业务系统的关键动作，例如查订单、改状态、发短信，不能指望模型自己编。函数调用（Function Calling / Tool Use）是解决这个问题的标准手段：模型输出一个结构化指令，告诉系统“该调哪个函数，参数是什么”，真正的执行还是由代码完成。

下面是一个函数定义片段：

tools = [ { "type": "function", "function": { "name": "query_order", "description": "根据订单号查询用户订单详情", "parameters": { "type": "object", "properties": { "order_id": { "type": "string", "description": "订单号，例如SO20240101" } }, "required": ["order_id"] } } } ]

调用时，把tools列表连同用户消息一起发给模型。模型会根据内容决定是否需要调用工具。如果需要，它的返回里会带一个结构化的调用请求，而不是自然语言回复。系统按这个请求去执行代码，再把结果追加到对话里让模型继续生成。这个循环就是Agent的基础。

我在实际项目里会特别注意一点：函数的描述（description）写得越细，模型越不会乱用。描述里要包含“什么时候用”“参数从哪里取”。如果你不写清楚，模型可能会把不存在的时间字符串当参数传进去，然后系统再去查一次空结果，白白增加一次无用调用。

3. 从一个Agent项目落地AI工程

3.1 场景定义：做一个工单分类+回复助手

纸上谈兵再多，不如跑一遍。我们来看一个最小但完整的Agent项目：工单分类与回复助手。整体流程是：用户提交工单 -> 系统提取工单信息 -> 模型判断是否需要查询订单 -> 如果需要则调用查询函数 -> 拿到结果后生成分类和回复。

项目虽然小，但它具备了AI工程的大部分要素：工具调用、上下文管理、缓存、对外服务接口。为了方便演示，代码环境假设为Python 3.10+，使用requests库直接和模型API交互。所有密钥通过环境变量读取，不写死在代码里。

3.2 搭环境与主流程代码

先安装必要的库：

pip install fastapi uvicorn requests redis

然后写模型调用封装。每个厂商的接口路径都会不同，这里以DashScope的兼容接口为例，但整体调用的主逻辑可以复用到其他服务。

import os import json import requests API_KEY = os.getenv("DASHSCOPE_API_KEY") MODEL_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" def call_llm(messages, tools=None, temperature=0.1): headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } payload = { "model": "qwen-plus", "messages": messages, "temperature": temperature } if tools: payload["tools"] = tools resp = requests.post(MODEL_URL, headers=headers, json=payload, timeout=60) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]

接下来定义工单查询函数，这里模拟查询数据库：

def query_order(order_id: str) -> dict: # 真实场景这里会执行SQL或调用内部API fake_db = { "SO20240101": {"status": "已发货", "item": "智能门锁", "logistics": "顺丰速运"}, "SO20240102": {"status": "待付款", "item": "智能摄像头", "logistics": ""} } order = fake_db.get(order_id) if not order: return {"error": "订单不存在"} return order

然后实现Agent主循环。核心思路就是：先让模型看工单内容，如果模型返回函数调用指令，就执行函数，把结果追加进messages，再让模型继续生成最终答案。

def agent_run(ticket_text: str): messages = [ {"role": "system", "content": "你是工单分类助手，必须通过查询订单后再回答用户问题。"}, {"role": "user", "content": f"请处理以下工单：{ticket_text}"} ] tool_map = { "query_order": query_order } for _ in range(3): # 最多循环3轮，防止无限调用 message = call_llm(messages, tools=tools) messages.append(message) if not message.get("tool_calls"): return message["content"] for tool_call in message["tool_calls"]: fn_name = tool_call["function"]["name"] args = json.loads(tool_call["function"]["arguments"]) result = tool_map[fn_name](**args) messages.append({ "role": "tool", "tool_call_id": tool_call["id"], "content": json.dumps(result, ensure_ascii=False) }) return "处理失败，超过最大循环次数"

这段代码最核心的地方在于“把工具执行结果回传给模型”。如果你漏掉这一环节，模型就会自己脑补订单状态，这也是很多AI Agent在真实场景里随口乱编的根源。每次工具调用后，都要带着真实结果回填到消息历史里，模型才知道接下来该怎么作答。

3.3 加缓存与服务化

生产环境不能每来一个请求都重新调模型。同一个用户的同一类问题，重复请求会造成成本浪费。我一般会加两层缓存：一层是对重复工单直接命中历史结果，另一层是对模型生成结果做短时缓存。

Redis缓存实现很直观：

import redis r = redis.Redis(host="localhost", port=6379, decode_responses=True) def get_cache(key: str): return r.get(key) def set_cache(key: str, value: str, ttl=3600): r.set(key, value, ex=ttl)

在调用Agent前，先根据工单文本的哈希值查缓存，命中就直接返回。要注意的是，缓存粒度不能太粗，否则带敏感信息的查询结果被缓存，既浪费存储又有数据泄露风险。我建议只缓存最终的分类和回复文本，不缓存中间工具调用结果。

把上面的代码用FastAPI包一层，就是一个最小可用的AI服务：

from fastapi import FastAPI, Body app = FastAPI() @app.post("/api/v1/ticket-agent") def handle_ticket(req: dict = Body(...)): ticket_text = req.get("ticket_text", "") cache_key = f"ticket:{abs(hash(ticket_text))}" cached = get_cache(cache_key) if cached: return {"content": cached, "source": "cache"} content = agent_run(ticket_text) set_cache(cache_key, content) return {"content": content, "source": "llm"}

到这里，一个Agent已经从“脚本”变成了“服务”。但先别急着上线，你还需要记录日志、加限流、做监控。我习惯在Agent入口和每次工具调用前后都打一条结构化日志，包含请求ID、模型调用耗时、token消耗、工具名称、执行结果。没有这些，后面排查问题就等于瞎猜。

4. 常见问题与排查技巧实录

4.1 API调用失败、超时和限流

真实环境中，模型接口不可能永远稳定。我遇到最多的是突然的限流和单次超时。限流时需要做指数退避重试，不能一失败就立刻重试，否则只会把服务打得更死。一个简单的退避间隔是：第一次等0.5秒，第二次1秒，第三次2秒，最多五次。

超时设置也要分层。网络连接超时要短一点，比如5秒；单次生成超时给到60秒甚至更长。因为长文本生成的耗时波动非常大，不能因为一次尾巴慢就杀掉整个请求。如果你用同步请求处理高并发，记得把线程池跑满，或者改用异步客户端。

4.2 模型不按JSON格式输出怎么办

这是Prompt工程里最常见的问题。即使你已经在模板里写了“必须输出JSON”，模型偶尔还是会给你夹带Markdown代码块、或者多个JSON对象粘在一起。

我的处理方式分三层。第一层是在Prompt里减少自由发挥，把输出格式用代码块圈起来。第二层是在解析时做容错，先把Markdown代码块去掉，再提取JSON片段。第三层是重试机制：如果解析失败，把“上次输出不符合格式要求”的错误信息回传给模型，让它重新生成。这一招在实际项目中能把成功率从92%拉到99%以上。

下面是一个容错解析的基础函数：

import re def extract_json(text: str): text = re.sub(r"```json|```", "", text).strip() start = text.find("{") end = text.rfind("}") if start == -1 or end == -1: raise ValueError("no json found") return json.loads(text[start:end+1])

4.3 Token超限如何拆分

长文本场景很容易触发上下文长度限制。很多人的第一反应是硬上更大的模型，但这不是解决根本问题。我建议使用滑动窗口策略：如果工单描述很长，先行做关键信息抽取，保留核心内容，去掉无意义的铺垫。

比如一个用户写了上千字的投诉，里面真正有用的可能就是最后三句话。我会先用一个较弱的模型或正则规则把时间、订单号、问题类型、期望结果抽取出来，再把结构化摘要喂给主模型。这样既控制token，又保证关键信息不丢。

还有一种情况是工具返回内容太长，比如查询订单返回了巨大的列表。这时应该对工具结果做摘要，只返回必要字段。我踩过坑，直接把列表全量塞给模型，结果一个request就爆掉，后来改成最多传10条记录，问题立刻缓解。

4.4 幻觉和上下文污染

幻觉不是模型“笨”，很多时候是Prompt给了它发挥空间。例如你问“订单SO99999状态如何”，模型没查到，但它为了让回答体面，会编一个“已签收”。解决办法有两个：一是要求模型在查到不到数据时明确说“没有查询到该订单”，二是把工具调用结果强制设置为唯一事实来源。

上下文污染指的是上一轮的输出错误地成为下一轮的输入。我在多轮Agent中经常看到这个问题，比如把模型生成的“回复建议”当成“用户消息”再次喂给模型，导致行为越来越怪。解决方法就是把消息角色严格区分：用户消息、系统消息、工具消息，每一轮追加都检查角色，不能混乱。

5. 从单Agent到多AI协作

5.1 多Agent的分工与协作模式

当任务复杂到单Agent难以独立完成时，就要考虑多AI协作。比如一个工单系统可以让分类Agent、质检Agent、回复Agent各司其职。分类Agent只负责给出类别；质检Agent检查用户的原始诉求是否被覆盖；回复Agent根据前面结果生成最终文案。

多Agent协作最怕的是互相甩锅。因为每个Agent的输入输出都没有统一格式，下游Agent拿到上游的垃圾输出后，会继续放大问题。我建议所有Agent之间通过结构化的消息对象传递，而不是直接传自然语言。每个消息对象至少包含：agent_id、task、status、payload、error。这样任何一个环节出问题，都能快速定位到具体模块。