“我把自己公司80%的业务交给了AI”这句话如果出现在朋友圈讨论大概率会滑向“AI会不会取代人”的情绪但如果它出现在一次技术评审会上就应该被拆成更尖锐的问题80%是按订单量、工单量还是工作时长算的AI 是给建议、做执行还是做最终决策答错了会不会引发客诉或合规问题出错以后是优化模型还是回滚流程我的判断是80% 不是 AI 能力参数而是一个管理流程参数。一家公司能把多少业务真正交出去首先取决于它对业务有多少结构化理解其次才是模型多强、Agent 多聪明。很多人做企业级 AI 落地感到卡壳不是因为模型不够好而是因为公司业务流程从来不是为了机器执行而设计的。哪些环节需要审批、哪些话术不能碰、哪些判断必须由人兜底这些隐性经验不被显性化AI 就算接进来也会一个接一个地出错。这篇文章把这个标题当成工程命题来处理。我会先讲清楚“交给 AI”到底意味着什么然后给出一套能支撑大规模业务接管的 AI 工程底座再用“客服 工单 知识库”的最小示例把链路跑通最后总结真实项目里最常踩的坑。读完你会得到一个可以带回去用的判断框架哪些业务该交、怎么交、怎么验证、以及怎么在出错时安全收回。1. 先把“80%”从口号翻译成一个工程目标如果业务负责人直接扔来一句话“我们要把 80% 的业务交给 AI”技术负责人第一时间不应该反驳也不应该马上承诺而是要把这句话翻译成可度量、可验证的工程目标。第一件事是选定业务范围。80% 必须落在一个具体场景上比如“客服一线可以独立处理的咨询量占比达到 80%”或者“标准工单的分类准确率达到 80%”。脱离业务范围谈比例约等于不做需求分析就估工作量。第二件事是把“交给 AI”改成行为标准AI 能直接给出可下发的答案人在置信度高于阈值时只做抽检低于阈值时必须介入。只有把口号转成“接管边界”后面才能设计灰度方案和回滚策略。下一步要区分业务的可自动化程度。不是所有业务都值得用 AI Agent 重构更不是所有业务都应该让模型做决定。参考下面这个分层方式业务类型典型场景建议处理方式可直接自动化工单分类、物流状态查询、报表初稿、文档摘要、数据格式转换第一批试点AI 独立完成人工抽检人机协同客诉回复、代码审查、合同初审、方案撰写AI 生成初稿人做判断和确认暂不建议独立执行涉及资产处分、对外法律承诺、核心产品方向、重大财务操作AI 只做辅助分析最终决策必须留给人这里真正容易踩坑的地方是把“辅助型业务”误当成“自动型业务”。比如同样是用大模型写客诉回复如果这家公司对客诉话术有强制边界比如不能主动承诺赔偿比例那就不能让它直接发出去必须经过审核节点。这个审核节点不是临时加的而是在业务分层时就该画进去的。另外要提醒一点80% 这个数字可能来自“AI 可以完成很多任务”的体感但体感不等于可接管率。一个任务能被 AI 接管前提是它有相对稳定的输入输出结构并且有可验证的验收标准。聊天机器人能聊 100 个话题不代表它能替公司处理 100 个业务真正决定接管率的是业务本身有多少“确定性”。2. “交给 AI”的本质是把业务流程重新形式化很多团队以为“把业务交给 AI”就是把大模型接入企业微信或者客服系统剩下的事情交给模型自由发挥。这个理解偏差是大量 AI 项目从演示到落地之间断层的主要原因。传统软件系统里业务逻辑是用代码写死的状态机、规则引擎、数据库事务、人工审批流。每一件事都有明确的触发条件、执行动作和结果分支。而大模型出现之后很多人希望跳过这套确定性工程直接让模型理解业务。问题是模型只能从上下文里猜出业务规则它不会自动知道你公司的订单改签流程需要先核验身份、再检查库存、然后走风控。模型一旦猜错错误会沿着后续链路被放大。“交给 AI”的正确姿势是把过去藏在人脑里的隐性经验重新表达成机器能执行的业务链路。具体说需要完成四步。第一步把业务拆到“事件—动作—判断”的粒度。不要只说“处理用户咨询”而要写清楚用户来咨询时系统先判断咨询类型再决定是查订单、查知识库还是转人工查完订单后什么状态下可以直接回复什么状态下必须转给售后。第二步把输入输出结构化。用户在对话框里说的话是自然语言但业务系统需要结构化字段。Agent 要先把自然语言转成 order_id、user_id、intent、sentiment 这类字段才能去调用订单接口。没有这一层结构化转换模型接手的就只是一堆文本后续工具根本无法可靠执行。第三步为每个动作定义验收标准和置信度阈值。AI 查到了订单状态接下来要判断它给出的“已发货”结论可不可以直接告诉用户。如果内部知识库显示这条物流信息已经超过 7 天未更新是否应该转人工确认这类验收规则需要在系统设计阶段就定好而不是等模型回答了再靠人肉检查。第四步设计人工兜底和数据回流机制。AI 一定会遇到不知道答案、知识库里没有对应内容、或者用户情绪激烈的情况。系统必须能主动“承认不知道”把会话转给人工而不是硬编一个答案。转人工之后人工改了什么、补了什么这些数据要回流到知识库和提示词模板里否则系统不会越用越准。这也是为什么很多公司“先买大模型再找场景”的做法容易失败。更稳妥的路径是先拆流程再选模型。当一条业务能被清晰地表达成事件、判断、动作和兜底规则时接大模型只是最后一步当业务本身还混沌不清时换再强的模型也救不了流程缺陷。3. 支撑“80%业务”的 AI 工程底座五个层次要稳定承担高比例业务AI 系统不能只有一个聊天窗口。从工程架构角度看支撑大规模接管的底座至少需要五层。第一层是接入层负责接收业务事件。客服消息、工单创建、邮件、内部审批请求都是系统入口。这一层需要考虑的不是模型而是消息队列、API 网关、鉴权以及如何把外部事件标准化成内部统一的消息结构。第二层是语义路由层负责理解用户到底想做什么。这一层通常是模型发挥核心价值的地方意图识别、实体抽取、情绪判断、是否需要转人工。路由层的输出必须是结构化的 JSON而不是一句“用户可能想问订单问题”的自然语言。第三层是执行层负责调用真实业务工具。比如查订单系统的发货状态、在工单系统里创建工单、给客户发送通知。这里的关键设计是Agent 不应该直接连数据库或核心系统而是应该通过统一工具网关调用 API。工具网关负责鉴权、参数校验、操作限流并把执行结果标准化。第四层是知识与记忆层。企业有产品手册、售后政策、历史工单、SOP 文档。这些内容要先做清洗、拆分、向量化再放进知识库供模型检索引用。没有知识层模型只能用通用知识回答回答得再流畅也不是企业要的业务答案。第五层是治理与兜底层。包括操作审计、人工队列、成本监控、版本回滚、权限控制。这一层决定了系统能不能长期运行。模型可以允许偶尔回答不够漂亮但不能允许越权操作没有日志、不能允许人工兜底队列无人响应。Agent 在这套架构里承担什么角色它更像一个“会使用工具的业务流程执行器”模型负责从用户消息中理解意图、拆解步骤但每一步工具调用都要受规则约束。如果在架构上把 Agent 当成一个万能数字员工期望它自己搞定所有事情系统会变得不可控。工程上更稳妥的做法是把 Agent 当成一个“不确定性决策器”在它外围架设确定性的护栏。从一次请求的视角看完整链路是这样的用户消息 → API 网关接入 → 语义路由识别意图、抽取实体、评估风险 → 知识检索召回相关内部文档片段 → Agent 编排决定调用哪些工具、按什么顺序调用 - 业务 API 执行查订单 / 创建工单 / 发送通知 → 汇总答案并计算置信度 → 低置信度或高风险时转人工队列 → 审计日志落库这个链路本身并不复杂但每一条线上都可能出问题。路由分错后面全错知识检索不准答案就缺乏依据工具权限过大一次错误调用可能影响真实业务数据兜底不及时转人工之后客户照样投诉。4. 环境准备与最小技术栈选型如果你准备在公司内部做一个最小闭环验证不需要一开始就打造完美的 AI 中台。先选一条业务线用最朴素的技术栈把链路跑通比买一堆平台更重要。下面这组选型适合多数中小型技术团队快速做概念验证但不代表生产环境唯一标准。编程语言Python 3.10适合做原型如果你所在团队是 Java 技术栈也可以用 Spring AI 或 LangChain4j核心思路一样。模型接入方式使用 OpenAI 兼容协议的模型网关。不管底层用闭源模型还是开源模型统一走 chat/completions 和 embeddings 接口方便后续替换模型。存储与检索PostgreSQL pgvector。它同时能存业务数据、审计日志和向量索引适合中小规模场景大规模场景再考虑独立向量数据库。消息与兜底队列先用 Redis List 或 PostgreSQL 表模拟跑通后再接 RabbitMQ 或 Kafka。版本管理配置、提示词、知识库文档都要纳入 Git 管理AI 工程同样需要回滚能力。需要说明的是大模型版本、数据库版本和依赖库版本请以你实际安装的官方文档为准。下面示例的重点是通用实现思路不是绑定某一家厂商。动手之前建议先做一个环境自检清单模型网关地址能不能通过 curl 调用鉴权方式是什么。是否已经准备好有权限访问的测试订单接口或工单系统接口。PostgreSQL 是否已经安装 pgvector 插件测试库里能否执行向量检索语句。是否已经准备好了第一批内部知识文档而不是拿互联网上的通用文档做测试。是否明确人工兜底队列由谁在什么时间范围内响应。这些前置条件看起来琐碎但决定了一个 AI Agent 示例能不能在两周内真正落地。跳过这些检查项目大概率会卡在“模型能聊天但不能干正事”的阶段。5. 完整示例一个“客服工单知识库”的自动化闭环下面用一个最小但完整的示例演示链路。场景设定为客服收到用户消息后系统判断是订单查询、产品知识还是投诉需要查知识库或订单接口的调用工具完成回答低置信度或投诉类型直接转人工队列所有操作记录审计日志。为了让示例尽量完整代码分成配置文件、语义路由、知识检索、审计日志和 Agent 主循环几部分。实际生产系统中建议补上链路追踪、权限网关和人工 SLA 管理但下面代码已经足够把一个最小闭环跑起来。5.1 配置文件文件路径config/business_agent.yamlllm: base_url: http://你的模型网关地址/v1 api_key_env: LLM_API_KEY model: 你的模型名称 router: temperature: 0 max_retries: 2 knowledge: top_k: 4 min_score: 0.35 agent: max_steps: 8 low_confidence_threshold: 0.65 fallback: queue: human_ticket reason: low_confidence_or_route_need_human这里把模型地址、知识检索参数和置信度阈值放在配置里是为了让非开发同事也能调整。真正容易踩坑的是 threshold 这个参数。调太高大量请求会转人工AI 接管率上不来调太低错误答案会直接发出去。建议先设一个偏保守的值比如 0.75再根据实际数据慢慢下调到 0.6 左右但不要低于 0.5。5.2 语义路由模块文件路径agent/router.pyimport json import os import requests def load_cfg(path: str) - dict: import yaml with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def chat_once(messages: list, cfg: dict) - str: url cfg[llm][base_url].rstrip(/) /chat/completions headers { Authorization: Bearer os.environ[cfg[llm][api_key_env]] } payload { model: cfg[llm][model], messages: messages, temperature: cfg[router][temperature], } resp requests.post(url, headersheaders, jsonpayload, timeout30) resp.raise_for_status() return resp.json()[choices][0][message][content] def classify_ticket(text: str, cfg: dict) - dict: system_prompt 你是企业内部服务台的业务路由员。 请判断用户请求类型只输出 JSON字段为 - type: order_query(订单查询) / product_knowledge(产品知识) / complaint(投诉) / other(其他) - confidence: 0 到 1 之间的小数 - need_human: 是否需要人工处理布尔值 示例输出{type: order_query, confidence: 0.9, need_human: false} content chat_once([ {role: system, content: system_prompt}, {role: user, content: text} ], cfg) try: return json.loads(content) except json.JSONDecodeError: # 模型如果返回了多余文本这里做最简单清洗后再解析 start content.find({) end content.rfind(}) 1 return json.loads(content[start:end])这个模块的核心作用是把用户的一句话转化成路由判断和结构化字段。如果大模型在分类阶段就出错后续所有环节都会失效。所以在示例代码里我把温度设置为 0并明确要求模型只输出 JSON。如果你的模型网关支持 JSON Mode生产环境更推荐直接开启避免解析失败。5.3 知识检索模块文件路径agent/knowledge.pyimport os import requests import psycopg def get_embedding(text: str, cfg: dict) - list: url cfg[llm][base_url].rstrip(/) /embeddings headers { Authorization: Bearer os.environ[cfg[llm][api_key_env]] } payload { model: cfg[llm][model], input: text } resp requests.post(url, headersheaders, jsonpayload, timeout30) resp.raise_for_status() return resp.json()[data][0][embedding] def retrieve_top_k(query: str, cfg: dict) - list: embedding get_embedding(query, cfg) conn psycopg.connect(postgresql://user:passwordlocalhost:5432/business_db) rows conn.execute( SELECT content, 1 - (embedding %s::vector) AS score FROM knowledge_chunks WHERE is_latest_version true ORDER BY embedding %s::vector LIMIT %s , (embedding, embedding, cfg[knowledge][top_k]) ).fetchall() conn.close() return [{content: row[0], score: float(row[1])} for row in rows]这段代码做了两件关键事。第一把用户问题转换成向量第二从知识库里召回最相关的文档片段并且只查最新版本。注意 SQL 里的 is_latest_version 字段这是实际项目中必须有的设计。知识库里的政策会更新如果旧版本文档没有被过滤模型很可能引用过期内容给出违反公司当前规定的错误答案。关于 embedding 参数的传参方式不同 Python 数据库驱动要求不同有的需要转成字符串有的直接传数组请按你实际使用的 psycopg 版本文档调整。5.4 审计日志模块文件路径agent/audit.pyimport json import uuid import psycopg AUDIT_TABLE business_agent_audit_log def save_audit_log( user_query: str, route_type: str, final_answer: str, source_chunks: list, confidence: float, status: str, reason: str , handled_by: str ai, ): conn psycopg.connect(postgresql://user:passwordlocalhost:5432/business_db) request_id str(uuid.uuid4()) conn.execute( f INSERT INTO {AUDIT_TABLE} ( request_id, user_query, route_type, final_answer, source_chunks, confidence, handled_by, status, reason ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) , ( request_id, user_query, route_type, final_answer, json.dumps(source_chunks, ensure_asciiFalse), confidence, handled_by, status, reason, ), ) conn.commit() conn.close() return request_id很多 AI 项目只关心回答得准不准却忽略审计。没有审计日志出问题之后完全无法定位是哪一次的 prompt、哪一份知识库文档、哪一个模型判断导致了错误。审计日志是 Agent 系统里的“黑匣子”宁可多记不能少记。5.5 Agent 主循环文件路径agent/main_loop.pyimport json import sys from router import load_cfg, chat_once from knowledge import retrieve_top_k from audit import save_audit_log def run_agent(user_text: str, cfg: dict) - dict: route classify_ticket(user_text, cfg) route_type route.get(type, other) confidence float(route.get(confidence, 0.0)) need_human route.get(need
“我把自己公司80%的业务交给了AI”这句话如果出现在朋友圈讨论大概率会滑向“AI会不会取代人”的情绪但如果它出现在一次技术评审会上就应该被拆成更尖锐的问题80%是按订单量、工单量还是工作时长算的AI 是给建议、做执行还是做最终决策答错了会不会引发客诉或合规问题出错以后是优化模型还是回滚流程我的判断是80% 不是 AI 能力参数而是一个管理流程参数。一家公司能把多少业务真正交出去首先取决于它对业务有多少结构化理解其次才是模型多强、Agent 多聪明。很多人做企业级 AI 落地感到卡壳不是因为模型不够好而是因为公司业务流程从来不是为了机器执行而设计的。哪些环节需要审批、哪些话术不能碰、哪些判断必须由人兜底这些隐性经验不被显性化AI 就算接进来也会一个接一个地出错。这篇文章把这个标题当成工程命题来处理。我会先讲清楚“交给 AI”到底意味着什么然后给出一套能支撑大规模业务接管的 AI 工程底座再用“客服 工单 知识库”的最小示例把链路跑通最后总结真实项目里最常踩的坑。读完你会得到一个可以带回去用的判断框架哪些业务该交、怎么交、怎么验证、以及怎么在出错时安全收回。1. 先把“80%”从口号翻译成一个工程目标如果业务负责人直接扔来一句话“我们要把 80% 的业务交给 AI”技术负责人第一时间不应该反驳也不应该马上承诺而是要把这句话翻译成可度量、可验证的工程目标。第一件事是选定业务范围。80% 必须落在一个具体场景上比如“客服一线可以独立处理的咨询量占比达到 80%”或者“标准工单的分类准确率达到 80%”。脱离业务范围谈比例约等于不做需求分析就估工作量。第二件事是把“交给 AI”改成行为标准AI 能直接给出可下发的答案人在置信度高于阈值时只做抽检低于阈值时必须介入。只有把口号转成“接管边界”后面才能设计灰度方案和回滚策略。下一步要区分业务的可自动化程度。不是所有业务都值得用 AI Agent 重构更不是所有业务都应该让模型做决定。参考下面这个分层方式业务类型典型场景建议处理方式可直接自动化工单分类、物流状态查询、报表初稿、文档摘要、数据格式转换第一批试点AI 独立完成人工抽检人机协同客诉回复、代码审查、合同初审、方案撰写AI 生成初稿人做判断和确认暂不建议独立执行涉及资产处分、对外法律承诺、核心产品方向、重大财务操作AI 只做辅助分析最终决策必须留给人这里真正容易踩坑的地方是把“辅助型业务”误当成“自动型业务”。比如同样是用大模型写客诉回复如果这家公司对客诉话术有强制边界比如不能主动承诺赔偿比例那就不能让它直接发出去必须经过审核节点。这个审核节点不是临时加的而是在业务分层时就该画进去的。另外要提醒一点80% 这个数字可能来自“AI 可以完成很多任务”的体感但体感不等于可接管率。一个任务能被 AI 接管前提是它有相对稳定的输入输出结构并且有可验证的验收标准。聊天机器人能聊 100 个话题不代表它能替公司处理 100 个业务真正决定接管率的是业务本身有多少“确定性”。2. “交给 AI”的本质是把业务流程重新形式化很多团队以为“把业务交给 AI”就是把大模型接入企业微信或者客服系统剩下的事情交给模型自由发挥。这个理解偏差是大量 AI 项目从演示到落地之间断层的主要原因。传统软件系统里业务逻辑是用代码写死的状态机、规则引擎、数据库事务、人工审批流。每一件事都有明确的触发条件、执行动作和结果分支。而大模型出现之后很多人希望跳过这套确定性工程直接让模型理解业务。问题是模型只能从上下文里猜出业务规则它不会自动知道你公司的订单改签流程需要先核验身份、再检查库存、然后走风控。模型一旦猜错错误会沿着后续链路被放大。“交给 AI”的正确姿势是把过去藏在人脑里的隐性经验重新表达成机器能执行的业务链路。具体说需要完成四步。第一步把业务拆到“事件—动作—判断”的粒度。不要只说“处理用户咨询”而要写清楚用户来咨询时系统先判断咨询类型再决定是查订单、查知识库还是转人工查完订单后什么状态下可以直接回复什么状态下必须转给售后。第二步把输入输出结构化。用户在对话框里说的话是自然语言但业务系统需要结构化字段。Agent 要先把自然语言转成 order_id、user_id、intent、sentiment 这类字段才能去调用订单接口。没有这一层结构化转换模型接手的就只是一堆文本后续工具根本无法可靠执行。第三步为每个动作定义验收标准和置信度阈值。AI 查到了订单状态接下来要判断它给出的“已发货”结论可不可以直接告诉用户。如果内部知识库显示这条物流信息已经超过 7 天未更新是否应该转人工确认这类验收规则需要在系统设计阶段就定好而不是等模型回答了再靠人肉检查。第四步设计人工兜底和数据回流机制。AI 一定会遇到不知道答案、知识库里没有对应内容、或者用户情绪激烈的情况。系统必须能主动“承认不知道”把会话转给人工而不是硬编一个答案。转人工之后人工改了什么、补了什么这些数据要回流到知识库和提示词模板里否则系统不会越用越准。这也是为什么很多公司“先买大模型再找场景”的做法容易失败。更稳妥的路径是先拆流程再选模型。当一条业务能被清晰地表达成事件、判断、动作和兜底规则时接大模型只是最后一步当业务本身还混沌不清时换再强的模型也救不了流程缺陷。3. 支撑“80%业务”的 AI 工程底座五个层次要稳定承担高比例业务AI 系统不能只有一个聊天窗口。从工程架构角度看支撑大规模接管的底座至少需要五层。第一层是接入层负责接收业务事件。客服消息、工单创建、邮件、内部审批请求都是系统入口。这一层需要考虑的不是模型而是消息队列、API 网关、鉴权以及如何把外部事件标准化成内部统一的消息结构。第二层是语义路由层负责理解用户到底想做什么。这一层通常是模型发挥核心价值的地方意图识别、实体抽取、情绪判断、是否需要转人工。路由层的输出必须是结构化的 JSON而不是一句“用户可能想问订单问题”的自然语言。第三层是执行层负责调用真实业务工具。比如查订单系统的发货状态、在工单系统里创建工单、给客户发送通知。这里的关键设计是Agent 不应该直接连数据库或核心系统而是应该通过统一工具网关调用 API。工具网关负责鉴权、参数校验、操作限流并把执行结果标准化。第四层是知识与记忆层。企业有产品手册、售后政策、历史工单、SOP 文档。这些内容要先做清洗、拆分、向量化再放进知识库供模型检索引用。没有知识层模型只能用通用知识回答回答得再流畅也不是企业要的业务答案。第五层是治理与兜底层。包括操作审计、人工队列、成本监控、版本回滚、权限控制。这一层决定了系统能不能长期运行。模型可以允许偶尔回答不够漂亮但不能允许越权操作没有日志、不能允许人工兜底队列无人响应。Agent 在这套架构里承担什么角色它更像一个“会使用工具的业务流程执行器”模型负责从用户消息中理解意图、拆解步骤但每一步工具调用都要受规则约束。如果在架构上把 Agent 当成一个万能数字员工期望它自己搞定所有事情系统会变得不可控。工程上更稳妥的做法是把 Agent 当成一个“不确定性决策器”在它外围架设确定性的护栏。从一次请求的视角看完整链路是这样的用户消息 → API 网关接入 → 语义路由识别意图、抽取实体、评估风险 → 知识检索召回相关内部文档片段 → Agent 编排决定调用哪些工具、按什么顺序调用 - 业务 API 执行查订单 / 创建工单 / 发送通知 → 汇总答案并计算置信度 → 低置信度或高风险时转人工队列 → 审计日志落库这个链路本身并不复杂但每一条线上都可能出问题。路由分错后面全错知识检索不准答案就缺乏依据工具权限过大一次错误调用可能影响真实业务数据兜底不及时转人工之后客户照样投诉。4. 环境准备与最小技术栈选型如果你准备在公司内部做一个最小闭环验证不需要一开始就打造完美的 AI 中台。先选一条业务线用最朴素的技术栈把链路跑通比买一堆平台更重要。下面这组选型适合多数中小型技术团队快速做概念验证但不代表生产环境唯一标准。编程语言Python 3.10适合做原型如果你所在团队是 Java 技术栈也可以用 Spring AI 或 LangChain4j核心思路一样。模型接入方式使用 OpenAI 兼容协议的模型网关。不管底层用闭源模型还是开源模型统一走 chat/completions 和 embeddings 接口方便后续替换模型。存储与检索PostgreSQL pgvector。它同时能存业务数据、审计日志和向量索引适合中小规模场景大规模场景再考虑独立向量数据库。消息与兜底队列先用 Redis List 或 PostgreSQL 表模拟跑通后再接 RabbitMQ 或 Kafka。版本管理配置、提示词、知识库文档都要纳入 Git 管理AI 工程同样需要回滚能力。需要说明的是大模型版本、数据库版本和依赖库版本请以你实际安装的官方文档为准。下面示例的重点是通用实现思路不是绑定某一家厂商。动手之前建议先做一个环境自检清单模型网关地址能不能通过 curl 调用鉴权方式是什么。是否已经准备好有权限访问的测试订单接口或工单系统接口。PostgreSQL 是否已经安装 pgvector 插件测试库里能否执行向量检索语句。是否已经准备好了第一批内部知识文档而不是拿互联网上的通用文档做测试。是否明确人工兜底队列由谁在什么时间范围内响应。这些前置条件看起来琐碎但决定了一个 AI Agent 示例能不能在两周内真正落地。跳过这些检查项目大概率会卡在“模型能聊天但不能干正事”的阶段。5. 完整示例一个“客服工单知识库”的自动化闭环下面用一个最小但完整的示例演示链路。场景设定为客服收到用户消息后系统判断是订单查询、产品知识还是投诉需要查知识库或订单接口的调用工具完成回答低置信度或投诉类型直接转人工队列所有操作记录审计日志。为了让示例尽量完整代码分成配置文件、语义路由、知识检索、审计日志和 Agent 主循环几部分。实际生产系统中建议补上链路追踪、权限网关和人工 SLA 管理但下面代码已经足够把一个最小闭环跑起来。5.1 配置文件文件路径config/business_agent.yamlllm: base_url: http://你的模型网关地址/v1 api_key_env: LLM_API_KEY model: 你的模型名称 router: temperature: 0 max_retries: 2 knowledge: top_k: 4 min_score: 0.35 agent: max_steps: 8 low_confidence_threshold: 0.65 fallback: queue: human_ticket reason: low_confidence_or_route_need_human这里把模型地址、知识检索参数和置信度阈值放在配置里是为了让非开发同事也能调整。真正容易踩坑的是 threshold 这个参数。调太高大量请求会转人工AI 接管率上不来调太低错误答案会直接发出去。建议先设一个偏保守的值比如 0.75再根据实际数据慢慢下调到 0.6 左右但不要低于 0.5。5.2 语义路由模块文件路径agent/router.pyimport json import os import requests def load_cfg(path: str) - dict: import yaml with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def chat_once(messages: list, cfg: dict) - str: url cfg[llm][base_url].rstrip(/) /chat/completions headers { Authorization: Bearer os.environ[cfg[llm][api_key_env]] } payload { model: cfg[llm][model], messages: messages, temperature: cfg[router][temperature], } resp requests.post(url, headersheaders, jsonpayload, timeout30) resp.raise_for_status() return resp.json()[choices][0][message][content] def classify_ticket(text: str, cfg: dict) - dict: system_prompt 你是企业内部服务台的业务路由员。 请判断用户请求类型只输出 JSON字段为 - type: order_query(订单查询) / product_knowledge(产品知识) / complaint(投诉) / other(其他) - confidence: 0 到 1 之间的小数 - need_human: 是否需要人工处理布尔值 示例输出{type: order_query, confidence: 0.9, need_human: false} content chat_once([ {role: system, content: system_prompt}, {role: user, content: text} ], cfg) try: return json.loads(content) except json.JSONDecodeError: # 模型如果返回了多余文本这里做最简单清洗后再解析 start content.find({) end content.rfind(}) 1 return json.loads(content[start:end])这个模块的核心作用是把用户的一句话转化成路由判断和结构化字段。如果大模型在分类阶段就出错后续所有环节都会失效。所以在示例代码里我把温度设置为 0并明确要求模型只输出 JSON。如果你的模型网关支持 JSON Mode生产环境更推荐直接开启避免解析失败。5.3 知识检索模块文件路径agent/knowledge.pyimport os import requests import psycopg def get_embedding(text: str, cfg: dict) - list: url cfg[llm][base_url].rstrip(/) /embeddings headers { Authorization: Bearer os.environ[cfg[llm][api_key_env]] } payload { model: cfg[llm][model], input: text } resp requests.post(url, headersheaders, jsonpayload, timeout30) resp.raise_for_status() return resp.json()[data][0][embedding] def retrieve_top_k(query: str, cfg: dict) - list: embedding get_embedding(query, cfg) conn psycopg.connect(postgresql://user:passwordlocalhost:5432/business_db) rows conn.execute( SELECT content, 1 - (embedding %s::vector) AS score FROM knowledge_chunks WHERE is_latest_version true ORDER BY embedding %s::vector LIMIT %s , (embedding, embedding, cfg[knowledge][top_k]) ).fetchall() conn.close() return [{content: row[0], score: float(row[1])} for row in rows]这段代码做了两件关键事。第一把用户问题转换成向量第二从知识库里召回最相关的文档片段并且只查最新版本。注意 SQL 里的 is_latest_version 字段这是实际项目中必须有的设计。知识库里的政策会更新如果旧版本文档没有被过滤模型很可能引用过期内容给出违反公司当前规定的错误答案。关于 embedding 参数的传参方式不同 Python 数据库驱动要求不同有的需要转成字符串有的直接传数组请按你实际使用的 psycopg 版本文档调整。5.4 审计日志模块文件路径agent/audit.pyimport json import uuid import psycopg AUDIT_TABLE business_agent_audit_log def save_audit_log( user_query: str, route_type: str, final_answer: str, source_chunks: list, confidence: float, status: str, reason: str , handled_by: str ai, ): conn psycopg.connect(postgresql://user:passwordlocalhost:5432/business_db) request_id str(uuid.uuid4()) conn.execute( f INSERT INTO {AUDIT_TABLE} ( request_id, user_query, route_type, final_answer, source_chunks, confidence, handled_by, status, reason ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) , ( request_id, user_query, route_type, final_answer, json.dumps(source_chunks, ensure_asciiFalse), confidence, handled_by, status, reason, ), ) conn.commit() conn.close() return request_id很多 AI 项目只关心回答得准不准却忽略审计。没有审计日志出问题之后完全无法定位是哪一次的 prompt、哪一份知识库文档、哪一个模型判断导致了错误。审计日志是 Agent 系统里的“黑匣子”宁可多记不能少记。5.5 Agent 主循环文件路径agent/main_loop.pyimport json import sys from router import load_cfg, chat_once from knowledge import retrieve_top_k from audit import save_audit_log def run_agent(user_text: str, cfg: dict) - dict: route classify_ticket(user_text, cfg) route_type route.get(type, other) confidence float(route.get(confidence, 0.0)) need_human route.get(need
