1. 这不是写代码而是“造一个会思考的同事”——AI智能体开发到底在做什么很多人看到“AI智能体”四个字第一反应是科幻电影里那个能自主决策、开口说话、甚至带点小脾气的机器人。其实离我们最近的智能体可能就是你手机里那个能记住你常点哪家咖啡、主动提醒“今天降温该加件外套”的语音助手或是你公司内部那个不用翻文档、直接回答“上季度华东区退货率超标的三个SKU是什么原因分析报告在哪”的业务助理。它不靠预设脚本硬扛所有问题而是像一个刚入职三个月、但已经摸清公司流程、熟悉数据位置、知道该问谁要什么材料的新人——它有目标、会拆解、懂调用、能反思、还会迭代。“AI智能体的开发流程”说白了就是把这种“类人协作能力”从模糊设想变成可运行、可维护、可进化的数字系统的过程。它和传统软件开发最根本的区别在于你不是在写功能逻辑而是在设计一套“认知操作系统”。这个系统要能理解模糊指令比如“帮我梳理客户投诉里的高频情绪词”能判断当前该调用哪个工具是查CRM数据库还是调用NLP情感分析API还是生成一份可视化图表能在执行失败时换条路走比如数据库连不上就先用本地缓存数据做初步统计还能把这次经验记下来下次类似任务做得更快更准。我带过好几个团队落地过这类项目从电商客服辅助到制造业设备故障预判发现最大的误区就是把它当成“大模型几个API调用”的拼凑活。结果做出来的东西要么一问三不知要么答非所问要么卡在某个环节死循环。真正跑得稳的智能体背后是一整套严谨的工程化闭环从目标定义开始就拒绝“让它聪明一点”这种虚话必须明确“它要在什么场景下对谁解决哪类具体问题效果如何量化”。比如不是“提升客服效率”而是“将首次响应时间压缩至15秒内且90%以上的问题无需人工转接”。这个目标直接决定了你要给它配什么样的记忆模块、设定多细的工具调用权限、设计多深的自我反思层级。这个流程之所以现在被反复讨论并不是因为技术突然突破了而是企业真实痛点到了临界点大量重复性、跨系统、需上下文理解的脑力劳动正卡在“人太忙没空教AI”和“AI太笨学不会规则”之间。而智能体开发本质上是在搭建一座桥——一边是人的业务经验与判断直觉另一边是机器的执行速度与数据处理广度。接下来我会带你一层层拆开这座桥的每一块砖它怎么设计、怎么搭、怎么测、怎么防塌全是我在产线踩坑后总结出的实操细节。2. 智能体不是“堆模型”而是“搭认知骨架”——整体设计与思路拆解很多人一上来就想选最强的大模型仿佛模型参数量越大智能体就越“聪明”。这就像盖楼前先去挑最贵的钢筋却忘了地基打在哪、承重墙怎么布局。真正的起点永远是任务域建模——用一张清晰的图画出这个智能体要服务的具体业务场景里所有关键角色、数据流、决策点和约束条件。2.1 为什么必须先画“任务域地图”而不是急着写提示词我见过太多团队在没搞清业务脉络前就花两周时间精调GPT-4的system prompt结果上线后发现用户问“上个月流失客户的复购可能性”模型能生成一篇漂亮的分析报告但完全没调用真实的客户行为数据库所有结论都是基于通用知识的合理想象。问题出在哪不是模型不行是任务域没定义清楚——“复购可能性”这个指标在你们公司是由哪个系统计算依据哪些字段更新频率是多少有没有人工审核环节这些信息必须在设计阶段就固化进智能体的“认知边界”里。所以我的标准动作是拉上业务方、数据工程师、一线使用者一起画一张三层任务域地图表层用户可见层列出所有典型用户指令按高频、中频、低频分类。比如客服场景里“查订单状态”是高频“申请特殊退款”是中频“解释某项政策的历史变更”是低频。高频指令决定响应速度SLA中频指令决定工具链复杂度低频指令则考验知识库覆盖深度。中层系统交互层标注每条指令背后需要调用的内部系统CRM、ERP、BI平台、外部API物流查询、支付网关、以及需要读取的结构化/非结构化数据源工单文本、通话录音转录稿、产品说明书PDF。特别注意标出各系统的认证方式OAuth2.0API KeySAML单点登录和调用频次限制比如物流API每分钟最多10次请求。底层约束与规则层这是最容易被忽略、却最致命的一层。包括数据权限客服只能看自己负责的客户不能跨区域查询、合规红线金融场景中禁止生成投资建议、业务逻辑退货必须先确认库存再触发退款、以及兜底策略当所有工具都不可用时返回哪句标准话术并转人工。这张图一旦定稿后续所有技术选型都有了锚点。比如中层标出要频繁调用ERP的生产计划模块那工具调用框架就必须支持长连接和会话保持底层标出强合规要求那整个推理链就必须设计成可审计、可回溯的步骤日志不能只输出最终答案。2.2 “智能体架构”不是玄学而是四块确定性积木市面上讲智能体架构的文章动辄“感知-决策-执行-反思”四层听着高大上但落地时根本不知道每层该放啥。我把这套逻辑彻底具象化拆成四个必须独立实现、又紧密咬合的模块每个模块都有明确输入输出和验收标准目标解析器Goal Parser输入用户原始指令自然语言输出结构化任务描述JSON格式包含核心动词查询/生成/审批/预测、主语订单/客户/设备、关键约束时间范围/地域/状态、预期输出格式表格/文字摘要/图表链接关键设计点它不负责“理解”只负责“提取”。比如用户说“帮我看看张三最近三次下单都买了啥列个表”它必须精准识别出动词“查看”主语“张三的订单”次数“3”输出“表格”。所有模糊表述如“最近”、“都”必须在此模块转换为确定值如“最近过去30天”“都全部商品明细”。我通常用轻量级微调模型如Phi-3-3.8B做这件事因为它快、可控、错误模式稳定比用大模型做全链路解析更可靠。工具调度器Tool Orchestrator输入目标解析器输出的结构化任务输出按顺序执行的工具调用指令序列含参数、超时设置、重试策略关键设计点它像一个冷静的项目经理不关心工具内部怎么工作只关心“谁在什么时候该干什么”。比如任务要求“查张三订单分析商品关联性”它必须判断先调用CRM查订单耗时2s再把订单ID传给推荐引擎API做关联分析耗时5s如果推荐API超时则降级为用本地规则库匹配耗时100ms。这个模块的代码必须可配置、可热更新——业务规则变了改配置文件就行不用动代码。记忆协调器Memory Coordinator输入当前任务上下文 历史交互片段用户偏好、常用筛选条件、已确认的事实输出注入当前推理过程的上下文增强包Context Augmentation Package关键设计点它不是简单存聊天记录而是分层管理记忆。短期记忆本次对话内的实体指代如“它”指代上句提到的设备用LLM的上下文窗口中期记忆用户习惯如“王经理总要导出Excel”存在Redis里带TTL自动过期长期记忆公司制度、产品参数存在向量数据库但必须带来源标记和最后验证时间。最关键的是每次调用前它必须做一次“记忆可信度校验”——比如用户说“按上月标准”它得先查数据库确认“上月标准”当前是否有效而不是盲目信任历史记录。反思评估器Reflection Evaluator输入工具执行结果 用户反馈显式评价如“不对”或隐式信号如“再详细点”、“换个说法”输出本次任务的执行质量评分0-100 可操作的优化建议如“CRM接口平均延迟升至3.2s建议增加缓存”、“用户70%追问集中在价格解释需补充定价规则知识库”关键设计点它不追求“完美反思”而追求“可落地改进”。我坚持用规则引擎Drools而非大模型来做这事因为规则明确当用户连续两次说“没听懂”就触发术语解释流程当工具调用失败率15%就自动告警并降级当某类问题人工介入率突增就标记相关知识库条目待审核。所有建议必须指向具体责任人和行动项比如“请数据组在24小时内验证CRM订单状态字段的ETL延迟”。这四个模块构成了智能体的“认知骨架”。它们之间通过定义清晰的API契约通信任何模块都可以独立替换、升级、压测。这才是工程化的核心——把不确定性大模型的幻觉、网络抖动、数据漂移关进确定性的笼子里。3. 从蓝图到可运行核心细节解析与实操要点设计图再漂亮不落地就是废纸。我把智能体开发中最容易栽跟头的五个核心细节掰开揉碎讲透。这些不是理论而是我在三个不同行业项目里用真金白银试错换来的血泪经验。3.1 目标解析器别让“自然语言理解”毁掉整个链条很多团队把目标解析器做成一个黑盒大模型调用结果发现模型对“帮我查一下昨天的销售”能准确提取时间但对“帮我查一下上个周五的销售”就经常错判成“上周五”实际是7天前而非5天前。问题根源在于自然语言的时间表达本质是业务规则不是语言学问题。我的解决方案是“双轨制解析”规则轨道Rule Track用正则业务词典硬编码所有确定性表达。比如# 预定义时间词典 TIME_KEYWORDS { 今天: lambda: datetime.now().date(), 昨天: lambda: datetime.now().date() - timedelta(days1), 本周: lambda: (datetime.now().date() - timedelta(daysdatetime.now().weekday())), 上个周五: lambda: get_last_friday(), # 自定义函数精确计算 }所有出现在业务文档、客服话术、内部邮件里的固定时间表达必须穷举入库。这部分覆盖了85%以上的高频指令。模型轨道Model Track只处理规则无法覆盖的模糊表达如“旺季开始后”、“促销活动期间”。此时才调用微调后的Phi-3模型但输入必须经过严格清洗去掉语气词、标准化缩写“Q3”→“第三季度”、强制补全上下文“促销活动期间”→“2024年Q3大促活动期间”。模型输出不是最终结果而是多个候选解析带置信度由规则轨道做最终仲裁。提示在测试阶段必须用“对抗样本”暴击这个模块。比如故意输入“上上个周一的销售”看它是否报错而不是瞎猜输入“上个月的今天”看它是否意识到2月没有30号。我要求每个时间解析规则必须附带10个以上真实业务场景下的正反例测试用例通不过就重写。3.2 工具调度器API不是万能胶而是需要精心设计的“关节”调度器常犯的错误是把所有API当平等公民对待。但现实是查库存的API毫秒级响应调用外部征信API可能要等15秒而生成一份PDF报告的后台任务根本就是异步的。如果调度器不懂区分整个智能体就会卡死。我的实践是给每个工具打上三维度标签工具ID响应类型平均延迟失败降级方案crm_order_query同步120ms返回缓存数据TTL5minexternal_credit_check同步8.2s跳过此步标记“信用信息待补”report_generator异步N/A任务ID发送邮件通知附带下载链接调度器根据这些标签动态生成执行策略对同步短延迟工具500ms直接串行调用对同步长延迟工具2s启动并行调用但设置严格超时如3s超时即切降级方案对异步工具立即返回“任务已提交预计X分钟后邮件通知”并在后台轮询状态状态变更时主动推送。最关键的是所有工具调用必须封装成“带契约的函数”。比如crm_order_query的契约定义def crm_order_query( customer_id: str, date_range: tuple[date, date], # 必须是确定日期不接受最近 max_results: int 100 # 防止OOM ) - List[OrderItem]: 【前置校验】customer_id必须匹配CRM正则 ^C\d{8}$ 【后置校验】返回订单数不能超过max_results否则抛异常 【超时】硬性300ms超时抛TimeoutError 这个契约既是开发规范也是测试用例的源头。每次工具升级只要契约不变调度器代码零修改。3.3 记忆协调器别把“记住”当成“背诵”要分层分级管理新手常犯的错误是把所有对话历史塞进LLM上下文美其名曰“增强记忆”。结果模型越来越慢还总把三天前用户随口说的“我喜欢吃辣”当成当前订单的口味偏好。我的分层记忆架构如下瞬时记忆Session Memory仅存于当前HTTP请求生命周期存储本次对话的实体指代链如“它”→“刚才提到的服务器”。用Python字典实现无持久化请求结束即销毁。会话记忆Conversation Memory存于RedisKey为conv:{session_id}Value为有序列表每条记录包含时间戳、用户消息、系统回复、关键实体提取结果。TTL设为24小时自动过期。重点是每次写入前必须做实体消歧检测新消息中的“它”是否与历史中的某个实体匹配匹配则更新该实体的最新状态不匹配则新建。用户画像记忆User Profile Memory存于PostgreSQL表结构包含user_id,preference_key,preference_value,last_updated,source来源用户显式设置/系统推断/管理员配置。关键设计是source字段——当用户说“以后都用简体中文”source设为user_explicit权重最高当系统根据用户常访问页面推断“偏好技术文档”source设为system_inferred权重较低且带confidence_score。记忆协调器读取时按source权重和last_updated时间综合排序确保最可靠、最新的信息优先注入。知识库记忆Knowledge Base Memory存于向量数据库如Qdrant但绝不直接存原文。每条知识必须经过“三重加工”① 提取核心事实如“XX型号服务器保修期3年”② 标注适用场景scene: hardware_support③ 标注权威来源和时效source: official_manual_v2.3, valid_from: 2024-01-01。检索时不仅匹配语义更强制过滤scene和valid_from避免把过期政策当现行规则。注意所有记忆写入操作必须是原子的。我用Redis的MULTI/EXEC事务包裹会话记忆更新用PostgreSQL的INSERT ... ON CONFLICT DO UPDATE处理用户画像确保并发写入不丢数据。曾经有个项目因忘记加事务导致用户偏好被覆盖花了两天才从日志里还原。3.4 反思评估器让智能体“学会吃亏”而不是“假装聪明”很多团队的反思模块就是让大模型读一遍对话然后写个“本次服务很好”的总结。这毫无价值。真正的反思必须驱动可测量的改进。我的评估器采用“三层漏斗”机制基础层Automated Checks纯规则毫秒级完成。检查项包括工具调用是否全部成功失败率0%即告警响应时间是否超SLA如客服场景15s输出是否包含禁止词汇如“绝对”、“肯定”、“100%”等绝对化表述是否触发了兜底话术如“我暂时无法回答请联系人工”体验层User Signal Analysis分析用户隐式反馈。比如用户发送“”或“。。。”视为困惑信号用户重复提问同一问题相似度85%视为未理解用户在回复后立即发起新会话间隔30s视为不满意。价值层Business Impact Review对接业务系统验证结果有效性。例如客服场景调用CRM API确认智能体给出的解决方案是否真的关闭了对应工单销售场景调用BI系统确认智能体推荐的客户30天内是否产生了新订单。只有同时通过三层检查的任务才被标记为“成功”。任一层失败评估器就生成一条结构化改进项写入内部Jira系统自动分配给对应负责人。比如“CRM订单查询工具在14:00-15:00时段失败率升至22%建议检查数据库连接池配置”这条记录会直接创建为运维组的紧急任务。3.5 安全与合规不是加个防火墙而是把红线刻进每一行代码在金融、医疗、政务类项目中安全不是附加功能而是设计前提。我见过最危险的做法是把敏感数据过滤放在LLM输出后——相当于让模型先“看过”所有客户身份证号再让另一个程序把它们抹掉。幻觉和缓存会让这种做法形同虚设。我的“安全左移”实践输入净化Input Sanitization在目标解析器之前就部署正则过滤器拦截所有疑似PII个人身份信息的输入。比如用户输入“查张三138****1234的订单”过滤器会立即报错“检测到手机号为保护隐私请使用客户ID查询”。这个过滤器必须独立部署不依赖任何模型。上下文隔离Context Isolation每个工具调用都在独立的沙箱环境中执行。CRM工具的沙箱只允许访问orders和customers两张表且customers表自动脱敏id_card字段返回***phone字段返回138****1234。沙箱环境由数据库代理如ProxySQL实现规则写死在代理层应用代码无权绕过。输出审计Output Auditing所有最终回复必须经过审计模块签名。签名内容包括原始用户指令哈希、调用的工具链含版本号、关键数据字段的哈希如订单号、生成时间戳。这个签名存于区块链存证服务如蚂蚁链不可篡改。一旦发生纠纷可凭签名秒级追溯完整执行路径。人工熔断Human-in-the-loop对高风险操作如“冻结账户”、“修改合同金额”调度器必须强制插入人工审批节点。不是弹窗让用户点“确定”而是将请求推送到指定审批人企业微信审批人必须用生物识别指纹/人脸二次确认审批记录实时同步至审计链。这些不是锦上添花而是项目上线的准入门槛。我坚持一个原则任何安全措施如果不能在开发环境用单元测试覆盖就不算真正落地。比如PII过滤器必须有100%覆盖率的测试用例涵盖所有国内身份证、手机号、银行卡号的正则变体。4. 实操全过程从零搭建一个电商客服智能体现在我们把前面所有理念浓缩成一个可立即上手的实战案例为一家中型电商公司搭建一个处理“订单状态查询”的智能体。全程基于开源技术栈不依赖任何商业API所有代码均可在本地复现。4.1 环境准备与依赖安装我们选择极简但生产可用的技术组合Python 3.11 FastAPIWeb框架 LangChain编排 Qdrant向量库 PostgreSQL结构化数据 Redis缓存。所有组件均用Docker Compose一键拉起避免环境差异。# 创建项目目录 mkdir ecommerce-agent cd ecommerce-agent # 初始化docker-compose.yml cat docker-compose.yml EOF version: 3.8 services: # PostgreSQL - 存储订单、用户、产品数据 db: image: postgres:15 environment: POSTGRES_DB: ecommerce POSTGRES_USER: agent POSTGRES_PASSWORD: secret volumes: - ./data/db:/var/lib/postgresql/data ports: - 5432:5432 # Redis - 存储会话记忆和缓存 redis: image: redis:7-alpine command: redis-server --save 60 1 --loglevel warning volumes: - ./data/redis:/data ports: - 6379:6379 # Qdrant - 存储产品知识库向量化 qdrant: image: qdrant/qdrant volumes: - ./data/qdrant:/qdrant/storage ports: - 6333:6333 # 后端API服务 api: build: . ports: - 8000:8000 depends_on: - db - redis - qdrant environment: DB_URL: postgresql://agent:secretdb:5432/ecommerce REDIS_URL: redis://redis:6379/0 QDRANT_URL: http://qdrant:6333 EOF # 创建Dockerfile cat Dockerfile EOF FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0:8000, --port, 8000] EOF # 创建依赖文件 cat requirements.txt EOF fastapi0.110.0 uvicorn0.29.0 psycopg2-binary2.9.9 redis4.6.0 qdrant-client1.8.0 langchain0.1.15 langchain-community0.0.25 sentence-transformers2.2.2 pydantic2.7.1 SQLAlchemy2.0.29 EOF # 初始化数据库结构 cat init_db.sql EOF CREATE TABLE IF NOT EXISTS orders ( id SERIAL PRIMARY KEY, order_id VARCHAR(32) UNIQUE NOT NULL, customer_id VARCHAR(32) NOT NULL, status VARCHAR(20) NOT NULL CHECK (status IN (pending, shipped, delivered, cancelled)), created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), shipped_at TIMESTAMP WITH TIME ZONE, delivered_at TIMESTAMP WITH TIME ZONE ); CREATE TABLE IF NOT EXISTS customers ( id SERIAL PRIMARY KEY, customer_id VARCHAR(32) UNIQUE NOT NULL, name VARCHAR(100), phone VARCHAR(20) ); -- 插入测试数据 INSERT INTO customers (customer_id, name, phone) VALUES (C001, 张三, 138****1234), (C002, 李四, 159****5678); INSERT INTO orders (order_id, customer_id, status, created_at, shipped_at) VALUES (ORD-2024-001, C001, shipped, 2024-05-01 10:00:0000, 2024-05-02 14:30:0000), (ORD-2024-002, C002, delivered, 2024-05-03 09:15:0000, 2024-05-04 11:20:0000), (ORD-2024-003, C001, pending, 2024-05-05 16:45:0000, NULL); EOF # 启动服务 docker-compose up -d --build # 等待服务就绪后初始化数据库 docker-compose exec db psql -U agent -d ecommerce -f /app/init_db.sql这段脚本会在本地启动一个完整的电商数据环境。注意phone字段在数据库里是明文便于演示但在实际生产中必须在应用层加密存储且所有SQL查询都通过ORM的hybrid_property自动脱敏。4.2 构建目标解析器让智能体听懂“查订单”我们不调用大模型而是用LangChain的StructuredOutputParser 自定义正则构建一个轻量、确定、可测试的目标解析器。# parser.py from langchain.output_parsers import StructuredOutputParser, ResponseSchema from langchain.prompts import PromptTemplate import re from typing import Dict, Optional # 定义输出结构 response_schemas [ ResponseSchema(nameaction, description动作类型如query_order), ResponseSchema(nameorder_id, description订单ID如ORD-2024-001), ResponseSchema(namecustomer_id, description客户ID如C001), ResponseSchema(nametime_range, description时间范围如2024-05-01 to 2024-05-05), ] output_parser StructuredOutputParser.from_response_schemas(response_schemas) # 构建提示模板注意这里只是引导核心逻辑在正则 prompt_template 你是一个电商客服智能体的目标解析器。 请严格按JSON格式输出只包含以下字段action, order_id, customer_id, time_range。 不要添加任何额外说明或解释。 用户指令{input} {format_instructions} prompt PromptTemplate( templateprompt_template, input_variables[input], partial_variables{format_instructions: output_parser.get_format_instructions()} ) # 核心正则解析器主逻辑 class OrderQueryParser: def __init__(self): # 预编译正则提高性能 self.order_id_pattern re.compile(r(ORD-\d{4}-\d{3}), re.IGNORECASE) self.customer_id_pattern re.compile(r(C\d{3}), re.IGNORECASE) self.time_keywords { 今天: lambda: self._today(), 昨天: lambda: self._yesterday(), 最近: lambda: self._last_7_days(), } def parse(self, user_input: str) - Dict: # 步骤1提取订单ID order_id_match self.order_id_pattern.search(user_input) order_id order_id_match.group(1) if order_id_match else None # 步骤2提取客户ID customer_id_match self.customer_id_pattern.search(user_input) customer_id customer_id_match.group(1) if customer_id_match else None # 步骤3提取时间范围简化版实际项目需更复杂 time_range None for kw, func in self.time_keywords.items(): if kw in user_input: time_range func() break # 步骤4确定动作 action query_order if 取消 in user_input or 作废 in user_input: action cancel_order elif 发货 in user_input or 寄出 in user_input: action ship_order return { action: action, order_id: order_id, customer_id: customer_id, time_range: time_range } def _today(self): from datetime import datetime return datetime.now().strftime(%Y-%m-%d) def _yesterday(self): from datetime import datetime, timedelta return (datetime.now() - timedelta(days1)).strftime(%Y-%m-%d) def _last_7_days(self): from datetime import datetime, timedelta start datetime.now() - timedelta(days7) end datetime.now() return f{start.strftime(%Y-%m-%d)} to {end.strftime(%Y-%m-%d)} # 测试解析器 if __name__ __main__: parser OrderQueryParser() test_cases [ 查一下订单ORD-2024-001的状态, C001最近下的单有哪些, 昨天发货的订单, ] for case in test_cases: print(f输入: {case}) print(f输出: {parser.parse(case)}\n)运行测试你会看到输入: 查一下订单ORD-2024-001的状态 输出: {action: query_order, order_id: ORD-2024-001, customer_id: None, time_range: None} 输入: C001最近下的单有哪些 输出: {action: query_order, order_id: None, customer_id: C001, time_range: 2024-05-01 to 2024-05-08}这个解析器的优势在于100%可预测、100%可测试、100%免模型。你可以为每个业务场景编写50个真实用户语句作为测试用例确保覆盖率。这才是工程化的起点。4.3 开发工具调度器让智能体“会干活”调度器的核心是Tool抽象。我们为订单查询创建一个符合LangChain标准的工具# tools/order_tool.py from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Optional, Type import psycopg2 from psycopg2.extras import RealDictCursor class OrderQueryInput(BaseModel): order_id: Optional[str] Field(description订单ID如ORD-2024-001) customer_id: Optional[str] Field(description客户ID如C001) limit: int Field(default10, description最多返回多少条记录) class OrderQueryTool(BaseTool): name order_query description 查询订单状态。可按订单ID或客户ID查询。 args_schema: Type[BaseModel] OrderQueryInput def _run( self, order_id: Optional[str] None, customer_id: Optional[str] None, limit: int 10 ) - str: # 1. 连接数据库生产中应使用连接池 conn psycopg2.connect( hostdb, databaseecommerce, useragent, passwordsecret ) cursor conn.cursor(cursor_factoryRealDictCursor) try: # 2. 构建安全SQL防止SQL注入 if order_id: query SELECT * FROM orders WHERE order_id %s LIMIT %s params (order_id, limit) elif customer_id: query SELECT * FROM orders WHERE customer_id %s ORDER BY created_at DESC LIMIT %s params (customer_id, limit) else: return 错误必须提供订单ID或客户ID。 cursor.execute(query, params) rows cursor.fetchall() # 3. 格式化结果脱敏手机号 result [] for row in rows: # 从customers表获取客户姓名脱敏手机号 cursor.execute(SELECT name, phone FROM customers WHERE customer_id %s, (row[customer_id],)) cust_row cursor.fetchone() if cust_row: # 脱敏手机号138****1234 masked_phone cust_row[phone][:3] **** cust_row[phone][-4:] result.append({ order_id: row[order_id], customer_name: cust_row[name], customer_phone: masked_phone, status: row[status], created_at: row[created_at].strftime(%Y-%m-%d %H:%M) }) if not result: return 未找到符合条件的订单。 # 4. 生成自然语言回复 if len(result) 1: item result[0] return f订单{item[order_id]}状态为{item[status]}客户{item[customer_name]}下单
