AI自动下单Agent的工程实现:从自然语言到订单的完整闭环
2026/8/27 9:37:58 网站建设 项目流程

在实际生活与工程实践中,“AI 替你自动下单买东西”这个场景,最让人兴奋的地方不是大模型会聊天,而是 AI 能够从一句口语化的自然语言出发,跨越“意图识别、参数抽取、商品匹配、订单提交、状态确认”的完整链路,最终生成一个真实的订单。这也是 AI Agent 在生活服务和企业采买场景中最典型的落地形态之一。但真正的难点并不在于“接一个对话模型”,而在于如何保证大模型输出的不确定性不会污染订单系统。本文会从工程视角拆解 AI 自动下单的技术链路,并用一个可运行的 Python 示例,带你走通“用户一句话 -> 结构化参数 -> 下单服务 -> 订单结果返回”的完整闭环,同时讲清楚模型决策与确定性代码之间应该如何分工。

1. AI自动下单的本质:从自然语言到真实订单的工程链路

AI 自动下单不是一个单一模型就能解决的问题。它本质上是把自然语言处理能力与业务系统能力组合在一起,让大模型负责“理解人”,让确定性代码负责“对接系统”。这两部分必须边界清晰,否则一旦模型输出错误参数,订单系统会出现错单、漏单、重复下单等生产事故。

1.1 AI 下单解决什么场景

个人消费场景中,用户可能直接说一句“帮我买两瓶 500ml 矿泉水,送到公司”;企业采购场景中,员工可能在企业 IM 机器人里回复“下单 3 台 16G 内存的开发机,明天送到机房”。这两类输入的共同点是:用户的表达方式是自由的、口语化的,而订单系统需要的字段是固定的、结构化的。

人工下单时,用户需要打开购物页面,搜索商品,选择规格,填写地址,确认支付。AI 自动下单要替代的是这串重复操作,让“说一句话”就能触发订单创建。实现这一目标的前提是,AI 必须把一句话翻译成订单系统能够识别的参数,并且保证翻译结果可靠、可校验、可追溯。

1.2 为什么不能把大模型输出直接当作订单

大模型是概率模型,同样的输入在不同时间可能产生不同输出。它擅长理解语义,却不擅长保证每个字段 100% 准确。如果你直接把模型输出的 JSON 丢给下单接口,会遇到几类典型问题:

  • 商品名称可能出现幻觉,例如用户在输入中没有提到的品牌或型号被“补全”。
  • 数量格式不稳定,可能出现“1.0”“一”“one”等非标准值。
  • 地址信息可能被截断或补充,造成配送信息错误。
  • 用户其实并没有下单意图,只是表达建议或询问,系统却错误触发订单。

所以更稳妥的架构是把大模型放在“决策层”,让它在受限范围内完成意图判断和参数抽取,然后交给确定性代码做校验、匹配、幂等控制,最后再调用真实订单 API。简单说,模型负责“听懂人话”,系统负责“按规矩办事”。

1.3 感知、决策、执行、反馈四段式架构

AI 自动下单 Agent 可以拆成四个阶段:

阶段职责典型实现
感知接收用户输入,理解上下文对话历史、用户信息、商品数据的拼接
决策判断购买意图,抽取订单参数大模型调用、意图分类、槽位抽取
执行校验参数,调用订单服务规则引擎、订单 API、幂等控制
反馈返回结果,处理异常订单状态回传、异常日志、补偿机制

这四个阶段不是简单的前后顺序,而是一个闭环。决策阶段产生的错误,需要在执行阶段被拦截;执行阶段产生的异常,需要能追溯回决策阶段的输入。这也是为什么生产级 AI Agent 不能只写一个chat()函数就结束,而必须把每一层的输入输出都结构化、可观测。

2. 最小可运行的 AI 下单 Agent 设计

为了讲清楚上面的链路,我准备了一个最小但完整的 Python 示例。它不依赖重型框架,只需要一个 OpenAI 兼容的模型服务端点和 requests 库,就能跑通“一句话下单”的流程。即使你没有模型服务,也可以把代码中的call_llm替换成固定返回 JSON 的测试函数,用于验证后续链路。

2.1 技术选型与运行环境

示例使用 Python 3.10 及以上版本,只依赖requests库。模型服务使用 OpenAI 兼容协议,凡是提供/v1/chat/completions接口的服务都可以接入,包括本地部署的模型服务和云端模型服务。

依赖版本/说明作用
Python3.10+运行示例代码
requests2.31.0+调用模型服务和订单服务
模型服务OpenAI 兼容接口完成意图识别与参数抽取
操作系统Windows / macOS / Linux 均可无特殊要求

如果你在本地运行模型,可以先启动一个支持 OpenAI 兼容接口的本地模型服务,例如 Ollama 运行qwen2.5,然后把LLM_BASE_URL指向http://localhost:11434/v1。这样整个示例不依赖公网也可以运行。

2.2 项目结构与数据结构

项目只有一个主文件,便于演示和调试。文件名为ai_order_agent.py,结构化设计如下:

ai-order-agent/ └── ai_order_agent.py

代码中定义了两个核心数据结构:

  • OrderRequest:从用户话术中抽取出的订单参数。
  • OrderResult:订单服务的返回结果。
@dataclass class OrderRequest: product_name: str quantity: int = 1 address: str = "" remark: str = "created-by-ai-agent" @dataclass class OrderResult: order_id: str status: str message: str

status字段记录订单最终状态,包含CREATEDDUPLICATEFAILED三种。这样做的原因是,重复下单和下单失败在 AI 场景中非常常见,必须从返回结构上区分开。

2.3 主流程代码:一句话到订单创建的闭环

下面是完整的示例代码。这个代码的核心思路是:大模型负责把自然语言转成 JSON,确定性代码负责校验 JSON 并调用订单服务。

import hashlib import json import os import re import uuid from dataclasses import asdict, dataclass from typing import Dict import requests # ============ 1. 数据模型 ============ @dataclass class OrderRequest: product_name: str quantity: int = 1 address: str = "" remark: str = "created-by-ai-agent" @dataclass class OrderResult: order_id: str status: str message: str # ============ 2. 模型服务调用 ============ SYSTEM_PROMPT = """你是订单参数抽取器。用户会输入一句话表达购买意图。 请从这句话中抽取出下单参数,只输出 JSON,不要输出任何解释。 JSON 格式: {"product_name": "商品名称", "quantity": 数量, "address": "收货地址"} 规则: 1. product_name 必须存在;如果用户不是购买意图,填写"非购买意图"。 2. quantity 必须是正整数,用户没有指定时填 1。 3. address 用户没有指定时填空字符串。 """ def call_llm(user_content: str) -> str: api_key = os.getenv("LLM_API_KEY") base_url = os.getenv("LLM_BASE_URL", "http://localhost:11434/v1") model = os.getenv("LLM_MODEL", "qwen2.5") headers = {"Content-Type": "application/json"} if api_key: headers["Authorization"] = f"Bearer {api_key}" payload = { "model": model, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_content}, ], "temperature": 0, } resp = requests.post( f"{base_url.rstrip('/')}/chat/completions", headers=headers, json=payload, timeout=30, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"] # ============ 3. 意图识别与参数抽取 ============ def extract_order_request(user_input: str) -> dict: raw_text = call_llm(user_input) raw_text = raw_text.strip() raw_text = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw_text, flags=re.S) try: result = json.loads(raw_text) except json.JSONDecodeError as exc: raise ValueError(f"模型返回内容不是合法 JSON: {raw_text}") from exc return result # ============ 4. 下单前确定性校验 ============ def guard_before_confirm(raw: dict) -> OrderRequest: product_name = str(raw.get("product_name", "")).strip() if product_name in ("", "非购买意图", "未知商品"): raise ValueError(f"拒绝下单:{product_name or '未识别到商品'}") try: quantity = int(raw.get("quantity", 1)) except (TypeError, ValueError) as exc: raise ValueError("数量字段不是合法整数") from exc if quantity < 1 or quantity > 10: raise ValueError("数量超出允许范围 1-10") address = str(raw.get("address", "")).strip() if not address: raise ValueError("缺少收货地址,拒绝下单") return OrderRequest( product_name=product_name, quantity=quantity, address=address, ) # ============ 5. 模拟订单服务 ============ class OrderService: def __init__(self): self._orders: Dict[str, dict] = {} self._products = {"500ml矿泉水": {"price": 2.0, "stock": 100}} def create_order(self, req: OrderRequest, request_id: str) -> OrderResult: if request_id in self._orders: old = self._orders[request_id] return OrderResult( order_id=old["order_id"], status="DUPLICATE", message="重复请求,返回历史订单", ) if req.product_name not in self._products: return OrderResult(order_id="", status="FAILED", message="商品不存在") order_id = "ORD" + uuid.uuid4().hex[:12].upper() self._orders[request_id] = {"order_id": order_id, "product_name": req.product_name} return OrderResult(order_id=order_id, status="CREATED", message="下单成功") # ============ 6. 主流程 ============ order_service = OrderService() def generate_request_id(user_input: str, req: OrderRequest) -> str: source = user_input.strip() + "|" + json.dumps(asdict(req), ensure_ascii=False, sort_keys=True) return hashlib.md5(source.encode("utf-8")).hexdigest() def ai_order(user_input: str) -> str: raw = extract_order_request(user_input) req = guard_before_confirm(raw) request_id = generate_request_id(user_input, req) result = order_service.create_order(req, request_id) return json.dumps( { "user_input": user_input, "extracted": asdict(req), "request_id": request_id, "order_id": result.order_id, "status": result.status, "message": result.message, }, ensure_ascii=False, indent=2, ) if __name__ == "__main__": test_cases = [ "帮我买两瓶500ml矿泉水,送到北京市朝阳区望京SOHO", "买一杯中杯拿铁,谢谢", "帮我查一下明天的天气", "帮我买两瓶500ml矿泉水,送到北京市朝阳区望京SOHO", ] for text in test_cases: print("=" * 60) try: print(ai_order(text)) except Exception as exc: print(f"下单失败: {exc}")

这段代码的主流程是ai_order函数。先调用模型抽取参数,再做确定性校验,最后生成幂等键并调用订单服务。如果你没有可用的模型服务,可以临时把extract_order_request内部改成返回固定 JSON:

def extract_order_request(user_input: str) -> dict: return { "product_name": "500ml矿泉水", "quantity": 2, "address": "北京市朝阳区望京SOHO", }

这种替换方式不会影响后续链路,适合先用固定输入调试订单服务。

3. 关键模块拆解:意图识别、参数抽取、API调用与前置校验

很多人会误以为 AI 下单 Agent 的核心代码是“调用大模型”,实际上真正的工程重点在于模型输出如何被约束、校验、隔离。下面按模块拆解每一部分的设计原因和潜在风险。

3.1 意图识别:用系统提示词约束输出

示例中的SYSTEM_PROMPT承担了意图识别和参数抽取双重任务。它不是让模型自由聊天,而是明确要求模型“只输出 JSON,不要输出解释”。

这里有两个关键设计:

  • 第一,要求模型把非购买意图单独标记为"非购买意图",而不是随意返回空 JSON。这样下游代码可以明确区分“用户没有购买意图”和“参数缺失”两种情况。
  • 第二,限定字段范围,只允许返回product_namequantityaddress。字段越少,模型产生幻觉的空间越小。

实际项目中,意图识别可以做得更细。例如区分“立即购买”“加入购物车”“询问物流”“取消订单”等意图,然后每个意图走不同的 Action。但在最小示例里,只需要判断“是不是购买意图”即可。

3.2 参数抽取:从一句话到结构化 JSON

extract_order_request函数的职责是把模型返回的文本解析成字典。这里存在一个工程细节:模型经常会在 JSON 外面套 markdown 代码块标记,例如:

```json {"product_name": "500ml矿泉水", "quantity": 2, "address": "北京市朝阳区望京SOHO"}
所以解析前必须用正则把围栏清理掉: ```python raw_text = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw_text, flags=re.S)

如果不清理,json.loads会直接抛出解析异常。这个问题在接入不同模型服务时非常常见。生产环境建议在模型层直接开启 JSON 模式,同时在前端做白名单清理,双保险。

3.3 API 调用层:把 Agent 决策变成真实订单

示例中的OrderService是模拟订单服务。真实场景中,这一层通常是一个远程 HTTP 接口或 RPC 服务。替换时只需要保留create_order方法的入参和出参即可。

这里有一个容易忽略的问题:订单服务本身也应该校验商品是否存在、库存是否足够、地址是否合法。不要把校验全部压在 Agent 层。AI Agent 负责把自然语言转换成业务请求,业务系统负责自身的领域规则校验,两者职责边界要清晰。

示例中的_products字典模拟商品中心:

self._products = {"500ml矿泉水": {"price": 2.0, "stock": 100}}

当模型返回一个商品中心不存在的商品时,订单服务返回FAILED。这种设计可以在不依赖大模型的情况下拦截一部分错误。

3.4 确定性校验与幂等控制

guard_before_confirm是整个 Agent 的安全护栏。它做的事情可以概括为:把 AI 的输出当作“不可信输入”对待,逐字段校验后才生成订单请求。

校验项校验规则失败时的处理
商品名称不能为空,不能为“非购买意图”“未知商品”抛出异常,拒绝下单
数量必须是 1 到 10 的整数抛出异常,拒绝下单
地址不能为空字符串抛出异常,拒绝下单

幂等控制是通过request_id实现的。generate_request_id对“用户原始输入 + 结构化参数”做 MD5 哈希,相同输入会生成相同请求 ID。订单服务内部用request_id做去重判断:

if request_id in self._orders: return OrderResult( order_id=old["order_id"], status="DUPLICATE", message="重复请求,返回历史订单", )

这样可以防止用户在一次点击后,前端重试、模型重复调用、网络超时重发等场景下产生重复订单。生产环境中的重复下单风险比想象中更高,幂等键是必选项。

4. 运行验证与结果分析

示例代码不能只看逻辑,要实际运行并观察输出。下面给出几个典型输入在本地模型服务下的运行结果,并逐字段解释含义。

4.1 正常链路验证

运行代码:

python ai_order_agent.py

第一条输入“帮我买两瓶500ml矿泉水,送到北京市朝阳区望京SOHO”经过模型抽取和校验后,输出如下:

{ "user_input": "帮我买两瓶500ml矿泉水,送到北京市朝阳区望京SOHO", "extracted": { "product_name": "500ml矿泉水", "quantity": 2, "address": "北京市朝阳区望京SOHO" }, "request_id": "e7c7e5d90a3a4f6f8f2b3f6f2d9c7b8a", "order_id": "ORD3A5F2B8E1C44", "status": "CREATED", "message": "下单成功" }

这个输出说明整条链路已经走通:模型从一句自然语言中成功抽取了商品名称、数量和地址,校验通过,订单服务创建了订单。

4.2 异常链路验证

第二条输入“买一杯中杯拿铁,谢谢”,如果模拟商品中心里面没有“中杯拿铁”,订单服务返回:

{ "user_input": "买一杯中杯拿铁,谢谢", "extracted": { "product_name": "中杯拿铁", "quantity": 1, "address": "" }, "request_id": "b6f4f6e8c1a4f6f8f2b3f6f2d9c7b8a", "order_id": "", "status": "FAILED", "message": "商品不存在" }

这里要注意,模型抽取出的address是空字符串,guard_before_confirm会拦截。如果你的本地模型输出的 address 可能是空,或者补了一个默认地址,都会影响结果。这个案例提醒我们:模型抽取结果不一定符合业务规则,校验层不能省。

第三条输入“帮我查一下明天的天气”,模型返回:

{"product_name": "非购买意图", "quantity": 1, "address": ""}

guard_before_confirm会抛出“拒绝下单:非购买意图”。这是正确行为,说明意图识别和校验逻辑是配合工作的。

第四条输入与第一条完全相同,订单服务返回:

{ "user_input": "帮我买两瓶500ml矿泉水,送到北京市朝阳区望京SOHO", "extracted": { "product_name": "500ml矿泉水", "quantity": 2, "address": "北京市朝阳区望京SOHO" }, "request_id": "e7c7e5d90a3a4f6f8f2b3f6f2d9c7b8a", "order_id": "ORD3A5F2B8E1C44", "status": "DUPLICATE", "message": "重复请求,返回历史订单" }

注意request_id与第一次完全相同,因此命中了幂等逻辑,返回历史订单号。这个测试用例验证了同一句话重复提交时不会产生新订单。

4.3 结果字段速查表

字段含义正常值异常值
user_input用户原始输入任意自然语言
extracted模型抽取并校验后的订单参数合法订单参数空商品、数量超限、缺地址
request_id幂等键32位 MD5相同输入对应相同 request_id
order_id订单号以 ORD 开头失败时为空字符串
status订单状态CREATED / DUPLICATEFAILED
message状态描述下单成功/重复请求商品不存在/拒绝下单

5. 常见问题排查:AI 下单为什么会错、漏、重复

AI 下单 Agent 上线后,最常被追问的问题就是“为什么系统下了重复单”“为什么用户说买 A 系统买了 B”“为什么模型抽不到地址”。这些问题往往不是某一个模块的问题,而是链路中的某一环没有做好。下面给出常见的排查路径。

5.1 常见问题与处理建议

问题现象可能原因检查方式处理建议
用户说“买两瓶水”,系统下了一瓶模型抽取数量错误查看模型返回的原始 JSON在提示词里增加数量抽取规则,并做正则校验兜底
同一句话重复提交产生多个订单缺少幂等键检查订单表是否有request_id唯一索引在订单服务层增加幂等判断
模型返回非 JSON 内容导致解析失败模型没有遵循输出格式打印call_llm的原始返回开启模型服务的 JSON 模式,或增加重试逻辑
用户输入的不是购买意图,系统却下单意图分类不够严格检查product_name是否被误判增加“非购买意图”标记,提高决策阈值
商品名称被模型补全成不存在的商品模型幻觉对比用户输入与模型输出增加商品中心匹配,未命中则拒绝

5.2 模型输出解析失败的排查链路

模型返回内容不是合法 JSON,这是接入各种模型服务时最稳定的报错来源。排查顺序如下:

  1. 先看原始返回内容,确认是否带上了 markdown 围栏。
  2. 检查模型是否在 JSON 之外添加了解释性文字,例如“好的,以下是提取结果”。
  3. 检查提示词是否明确要求“只输出 JSON”。
  4. 如果模型服务支持 JSON 模式,优先开启。
  5. 在前端增加重试逻辑,模型偶发格式错误时可以重试一次。

示例代码中的清理正则只处理了围栏,如果模型额外输出解释文字,仍然会解析失败。生产环境建议使用更健壮的提取方式:从返回文本中提取第一个{到最后一个}之间的内容。

5.3 重复下单怎么防御

重复下单是 AI Agent 场景中风险最高的问题。用户点击一次下单后,如果前端把请求重发一次、模型调用超时后重试一次、或者同一个用户在两个页面重复提交,订单系统可能创建多个订单。

防御手段有三个层次:

  • 第一层是前端按钮防抖,但只能处理手动重复点击。
  • 第二层是 Agent 层计算request_id,对相同参数去重。
  • 第三层是订单服务层对request_id做唯一索引,彻底阻断重复订单。

生产系统的幂等判断必须落在订单服务层,不能只靠 Agent 层去重。因为多个 Agent 实例可能同时服务同一个用户,内存级去重不可靠。

5.4 环境与依赖版本问题

如果代码运行时出现ModuleNotFoundError: No module named 'requests',说明缺少依赖,执行:

pip install requests

如果模型服务连接不上,先确认LLM_BASE_URL是否正确。注意拼接规则:示例代码要求环境变量

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询