1. 这不是概念演示是银行、加密巨头和RPA龙头每天真正在跑的AI代理模式“Production-Ready AI Agents”这个短语最近被讲得太多多到让人怀疑它是不是又一个PPT里的热词。但如果你真去看过Bank of America的内部技术简报、Coinbase的工程博客或者UiPath去年发布的客户案例白皮书你会发现一件事他们没在聊“未来可能怎么做”而是在说“上个月我们用这个Agent处理了237万笔反欺诈工单”“上周这个Agent自动完成了89%的客户开户KYC初审”“过去90天这个Agent为全球27个区域的财务团队节省了11,400小时人工核对时间”。这不是实验室Demo是嵌在核心业务流里、扛着SLA服务等级协议、要过审计、能接真实数据库、会写日志、能告警、出问题能回滚的AI代理——也就是真正意义上的Production-Ready。我带团队落地过6个跨行业的AI Agent项目从保险理赔到供应链调度最深的体会是90%的失败不是因为模型不够强而是因为从第一天起就把它当成了“调用一次大模型API的小脚本”而不是一个需要工程化设计、可观测、可运维、可审计的软件系统。Bank of America不用LangChain写信贷审批AgentCoinbase不拿Llama-3直接连交易所API做风控决策UiPath更不会让一个未经沙箱隔离的Agent直接修改ERP主数据——他们用的是一套经过千锤百炼的模式语言Pattern Language就像建筑行业有“承重墙剪力墙框架结构”的组合逻辑AI Agent的生产级落地也有它自己的、可复用、可验证、可审计的8种基础模式。这8种模式不是理论推演出来的而是从三家机构公开披露的技术文档、专利申请、GitHub开源片段、技术大会演讲逐行抠出来的。比如Bank of America在2023年Q3财报附录里提到的“multi-stage validation loop for loan underwriting”对应的是本文的模式3分阶段校验环Staged Validation LoopCoinbase工程师在PyCon演讲中展示的“real-time anomaly triage with fallback to human-in-the-loop”就是模式6异常熔断人工兜底Anomaly Circuit Breaker Human FallbackUiPath在2024年产品路线图里强调的“agent-to-agent handoff with state persistence”则精准匹配模式7状态持久化代理交接Stateful Agent Handoff。这些不是命名游戏每个模式背后都绑定了明确的触发条件、数据契约、错误边界、监控指标和回滚路径。接下来我会把这8个模式掰开揉碎告诉你它们在真实系统里长什么样、为什么必须这么设计、踩过哪些坑、怎么抄作业。2. 为什么这8种模式能活下来——从“能跑通”到“敢上线”的底层逻辑2.1 模式存在的根本动因对抗AI不确定性三座大山所有想把AI Agent推进生产环境的团队最终都会撞上三堵墙。这8种模式本质上就是针对这三堵墙的工程化应答方案。第一堵墙叫输出不可控性Output Unpredictability。大模型的输出不是确定性函数同一个Prompt输入微小扰动比如日期格式多一个空格结果可能从“批准贷款”变成“拒绝并标记高风险”。Bank of America的信贷审批Agent如果只依赖单次LLM调用哪怕准确率标称95%剩下5%的误判也意味着每年数千万美元的坏账或合规处罚。所以他们的方案是绝不让LLM直接做终局决策。模式1结构化输出约束强制LLM只生成JSON Schema定义的字段模式2双模验证器让规则引擎和LLM各自打分只有两者置信度都超阈值才放行模式4外部知识锚定要求所有判断必须引用KB中编号为KB-2023-087的监管条款原文。这不是限制AI而是给它画出安全运行的轨道。第二堵墙是上下文脆弱性Context Fragility。Agent在处理一个复杂任务时往往要跨多个步骤、调用多个工具、读取多份文档。但LLM的上下文窗口有限且长文本推理容易“忘记”前面的关键约束。UiPath的财务对账Agent曾出现过这样的故障第一步从邮件提取发票号第二步查ERP获取金额第三步比对银行流水——但在第三步时模型“忘了”第一步提取的发票号转而用当前日期生成了一个假号码去查询导致对账失败并触发错误付款。他们的解法是模式5显式状态快照即每完成一步就把关键中间状态如invoice_id: INV-2024-7891, erp_amount: 14280.50序列化成带版本号的键值对存入Redis后续步骤必须显式读取该键而非依赖上下文记忆。这看起来笨重但换来的是100%的状态可追溯。第三堵墙是系统耦合风险System Coupling Risk。很多团队一上来就想让Agent“全知全能”查数据库、调API、发邮件、改文件……结果一个下游服务抖动整个Agent链路雪崩。Coinbase的交易风控Agent曾因此在一次AWS us-east-1区域故障中误将正常交易标记为洗钱暂停了37分钟的现货交易。他们后来重构为模式8松耦合工具编排即Agent本身只做决策所有工具调用由独立的Tool Orchestrator服务执行该服务自带重试策略、熔断阈值、降级开关比如当交易所API超时自动切到本地缓存的行情快照。Agent与工具之间只传递标准化的Input/Output Schema不共享任何连接、线程或内存。提示这三堵墙不是“等模型变强就能消失”的问题而是AI作为概率性组件嵌入确定性系统时必然存在的工程矛盾。回避它等于把生产环境当沙盒正视它才能设计出真正可靠的模式。2.2 模式选择的黄金法则按业务SLA倒推技术选型选哪种模式从来不是看哪个“听起来酷”而是看你的业务场景对哪项指标最敏感。我整理了一张决策表这是我和Bank of America架构师一起喝咖啡时画在餐巾纸上的业务场景特征最敏感SLA指标推荐主导模式关键设计要点Bank of America实例高合规要求如反洗钱、信贷审批零误判容忍False Positive Rate 0.001%模式3分阶段校验环每阶段设独立阈值任一阶段失败即终止全程留痕供审计KYC初筛先规则引擎过滤剔除92%明显不合格再LLM分析模糊材料最后人工复核剩余8%高实时性要求如交易风控、IoT告警端到端延迟 800msP95模式6异常熔断人工兜底熔断阈值设为响应时间300ms或置信度0.85熔断后自动转人工队列并推送摘要实时交易监控当检测到异常模式若LLM分析耗时超阈值立即触发人工审核通道同时返回“待确认”状态长流程协作如客户开户、保险理赔流程成功率 99.5%且失败可精确归因模式7状态持久化代理交接每个Agent交接点生成唯一trace_id状态存入带TTL的Key-Value存储失败时可从任意checkpoint重放开户流程身份核验Agent → 资料录入Agent → 合规审查Agent每步状态存Redis超时自动告警这张表的核心逻辑是用确定性机制规则、阈值、状态存储去包裹不确定性组件LLM。比如当业务要求“零误判”时你不能寄希望于LLM自己变得无比准确而应该设计一个环路让LLM只负责它最擅长的部分理解模糊文本把最危险的终局决策权交给确定性更强的规则引擎或人工。这就像汽车的安全气囊——不是指望司机永远不犯错而是为错误准备确定性的缓冲。2.3 模式不是孤立的而是可组合的“乐高积木”现实中没有哪个Agent只用一种模式。真正的生产系统是多个模式的嵌套与组合。以Coinbase的“新用户资产安全初始化”Agent为例它完整串联了5种模式启动层模式1结构化输出约束——用户注册后Agent首先生成一个严格Schema的security_setup_plan.json包含required_steps: [2fa_enrollment, whitelist_address_addition, withdrawal_limit_set]等字段杜绝自由发挥执行层模式8松耦合工具编排——每个required_step被发送给独立的Tool Orchestrator由其调用2FA服务、地址白名单API、风控限额服务Agent只接收标准化结果校验层模式2双模验证器——当whitelist_address_addition返回成功Agent不直接信任而是调用链上扫描工具验证该地址是否真在白名单合约中双源一致才标记完成容错层模式6异常熔断人工兜底——若链上扫描超时5s立即熔断将任务推入人工审核队列并附带原始交易哈希和超时日志状态层模式7状态持久化代理交接——整个流程的plan_id、各步骤status、last_updated_timestamp存入DynamoDB供客服系统实时查询进度。这种组合不是随意堆砌而是遵循一个清晰的分层原则越靠近用户交互层越强调结构化与可控越靠近执行层越强调解耦与韧性越靠近决策层越强调验证与可溯。理解这一点你才能避免“为了用模式而用模式”的陷阱。3. 八大生产级AI Agent模式详解从原理、结构到实操代码片段3.1 模式1结构化输出约束Structured Output Constraint核心思想不让LLM“自由创作”而是强制它在一个预定义的、机器可解析的JSON Schema内填空。这是所有生产级Agent的起点也是最简单却最有效的防线。为什么必须用自由文本输出无法被程序直接消费。想象一下你的Agent返回“好的我已经把发票金额设为一万四千二百八十块五毛。”——下游系统怎么知道这个数字是14280.50还是1428050而如果它返回{invoice_amount: 14280.50, currency: USD}一切就变得确定、可校验、可审计。Bank of America实践细节他们在信贷审批Agent中使用OpenAI的response_format{type: json_object}参数并配合一个自研的Schema Validator。这个Validator不仅检查JSON语法还做业务级校验比如loan_amount必须是正数且小于max_approved_amount从规则引擎获取的动态值credit_score必须在300-850区间。任何校验失败Agent立即返回错误码ERR_SCHEMA_VIOLATION而非尝试“修复”输出。实操代码片段Python OpenAI SDKfrom openai import OpenAI import json import re client OpenAI() def generate_structured_output(prompt: str, schema: dict) - dict: schema 示例: { type: object, properties: { decision: {type: string, enum: [APPROVE, REJECT, PENDING]}, confidence_score: {type: number, minimum: 0.0, maximum: 1.0}, reasoning_summary: {type: string, maxLength: 500} }, required: [decision, confidence_score] } try: response client.chat.completions.create( modelgpt-4-turbo, messages[{role: user, content: prompt}], response_format{type: json_object}, temperature0.1, # 降低随机性 max_tokens500 ) # 1. 基础JSON解析 raw_json json.loads(response.choices[0].message.content) # 2. 业务Schema校验关键 for required_key in schema.get(required, []): if required_key not in raw_json: raise ValueError(fMissing required field: {required_key}) # 3. 数值范围校验 if confidence_score in raw_json: score raw_json[confidence_score] if not (0.0 score 1.0): raise ValueError(fconfidence_score out of range: {score}) return raw_json except json.JSONDecodeError as e: # 记录原始LLM输出用于调试 log_error(fJSON Parse Error: {response.choices[0].message.content[:200]}) raise except ValueError as e: log_error(fSchema Validation Error: {e}) raise # 使用示例 prompt 基于以下客户信息信用分720年收入$120,000负债比35%请做出贷款审批决策。 schema { type: object, properties: { decision: {type: string, enum: [APPROVE, REJECT, PENDING]}, confidence_score: {type: number, minimum: 0.0, maximum: 1.0}, reasoning_summary: {type: string, maxLength: 500} }, required: [decision, confidence_score] } result generate_structured_output(prompt, schema) print(result) # 输出: {decision: APPROVE, confidence_score: 0.92, reasoning_summary: 信用分和收入均高于阈值...}实操心得不要省略业务级校验OpenAI的response_format只保证JSON语法正确不保证业务逻辑。必须自己写校验逻辑这是生产环境的底线。温度值temperature设为0.1或更低生产环境不是创意写作确定性优先于多样性。为每个Schema定义明确的required字段避免LLM“偷懒”省略关键信息。记录原始LLM输出当校验失败时原始输出是调试的唯一线索务必存入日志。3.2 模式2双模验证器Dual-Mode Validator核心思想让LLM和一个确定性规则引擎Rule Engine对同一输入独立打分只有两者结论一致且置信度都达标时才采纳结果。这是对抗LLM“一本正经胡说八道”的终极保险。为什么必须用LLM可能给出逻辑完美但事实错误的答案。例如在保险理赔中LLM可能根据病历描述“剧烈腹痛、发热”推理出“急性阑尾炎”并给出高置信度0.95——但它忽略了病历末尾手写的“已排除阑尾炎确诊为胃肠炎”。规则引擎则会机械地匹配关键词“已排除阑尾炎”直接返回0分。双模验证能捕获这种致命偏差。UiPath实践细节UiPath在其RPAAI平台中将此模式封装为ConfidenceGate组件。它要求规则引擎输出一个rule_score0-100LLM输出一个llm_score0-1两者加权平均权重可配置需大于阈值如0.82才通过。更重要的是它强制记录两者的原始输出供后续审计。在一次医疗报销审核中正是通过对比发现LLM的推理链遗漏了关键否定词从而推动了Prompt优化。实操代码片段模拟双模验证流程# 规则引擎简化版实际可能是Drools或自研DSL def rule_engine_eval(input_data: dict) - float: 输入: {symptoms: [腹痛, 发热], diagnosis_notes: 已排除阑尾炎确诊为胃肠炎} 输出: 规则得分 (0-100) notes input_data.get(diagnosis_notes, ) # 硬编码规则出现已排除且后面跟着疾病名则得0分 if 已排除 in notes and any(disease in notes for disease in [阑尾炎, appendicitis]): return 0.0 # 其他规则... return 85.0 # LLM评估使用上一节的structured_output函数 def llm_eval(input_data: dict) - dict: prompt f患者症状{input_data[symptoms]}。诊断备注{input_data[diagnosis_notes]}。请输出JSON{{diagnosis: ..., confidence_score: 0.x}} schema { type: object, properties: { diagnosis: {type: string}, confidence_score: {type: number, minimum: 0.0, maximum: 1.0} }, required: [diagnosis, confidence_score] } return generate_structured_output(prompt, schema) def dual_mode_validate(input_data: dict, rule_weight: float 0.4, llm_weight: float 0.6, threshold: float 0.82) - dict: 双模验证主函数 返回: {valid: bool, final_score: float, rule_score: float, llm_score: float, llm_output: dict} rule_score rule_engine_eval(input_data) / 100.0 # 归一化到0-1 llm_output llm_eval(input_data) llm_score llm_output[confidence_score] final_score rule_weight * rule_score llm_weight * llm_score is_valid (final_score threshold) and (rule_score 0.0) # 规则引擎得分为0是硬性拒绝 # 记录详细日志生产环境必需 log_audit({ input_data: input_data, rule_score: rule_score, llm_score: llm_score, final_score: final_score, is_valid: is_valid, llm_output: llm_output, timestamp: time.time() }) return { valid: is_valid, final_score: final_score, rule_score: rule_score, llm_score: llm_score, llm_output: llm_output } # 使用示例 input_data { symptoms: [腹痛, 发热], diagnosis_notes: 已排除阑尾炎确诊为胃肠炎 } result dual_mode_validate(input_data) print(result) # 输出: {valid: False, final_score: 0.34, rule_score: 0.0, llm_score: 0.95, ...} # 尽管LLM信心很高但规则引擎一票否决实操心得规则引擎不是摆设它的规则必须来自领域专家且定期更新。UiPath要求所有规则变更必须附带业务负责人签字的变更说明。权重不是拍脑袋Bank of America在反洗钱场景中将规则引擎权重设为0.7因为监管条款是刚性的而在客户服务场景中LLM权重提至0.8因为用户体验更看重自然度。“规则得分为0”是硬性熔断点这代表存在明确的业务禁忌如“已排除某病”此时无论LLM多自信都必须拒绝。审计日志必须包含原始输入和两个模型的原始输出这是应对监管检查的唯一证据。3.3 模式3分阶段校验环Staged Validation Loop核心思想将一个复杂的决策过程拆解为多个逻辑上递进、技术上隔离的阶段每个阶段有自己的输入、输出、校验规则和失败处理策略。失败不直接终止而是降级到下一阶段或人工。为什么必须用单点校验太脆弱。比如信贷审批如果只在最后一步校验“总负债率50%”那么前面所有关于收入、资产的分析都可能白费。分阶段校验则像工厂流水线每个工位都有质检员问题在源头就被拦截。Bank of America实践细节他们的贷款审批Agent采用经典的三阶段环Stage 1准入过滤纯规则引擎。检查身份证有效性、黑名单、基础资质如年龄18。99%的申请在此阶段被快速拒绝不消耗LLM资源。Stage 2材料分析LLM主导。分析上传的收入证明、银行流水PDF提取关键字段。输出必须通过模式1的结构化约束。Stage 3综合决策混合模式。将Stage 1的规则结果、Stage 2的LLM提取数据、以及实时征信API返回的数据输入一个轻量级XGBoost模型生成最终评分。此阶段也应用模式2的双模验证。实操代码片段阶段化流程控制器class StagedValidationLoop: def __init__(self): self.stages [ (stage_1_eligibility, self._stage_1_eligibility), (stage_2_document_analysis, self._stage_2_document_analysis), (stage_3_final_decision, self._stage_3_final_decision) ] self.state {} # 共享状态字典 def _stage_1_eligibility(self, input_data: dict) - dict: Stage 1: Rule-based eligibility check # 检查身份证号格式 id_number input_data.get(id_number, ) if not re.match(r^\d{17}[\dXx]$, id_number): return {status: REJECTED, reason: Invalid ID format, stage: stage_1} # 检查年龄 birth_year int(id_number[6:10]) age 2024 - birth_year if age 18 or age 70: return {status: REJECTED, reason: Age out of range, stage: stage_1} # 检查黑名单 if is_in_blacklist(id_number): return {status: REJECTED, reason: In blacklist, stage: stage_1} # 通过进入下一阶段 self.state[eligibility_passed] True return {status: PASSED, stage: stage_1} def _stage_2_document_analysis(self, input_data: dict) - dict: Stage 2: LLM-based document analysis if not self.state.get(eligibility_passed): return {status: SKIPPED, reason: Stage 1 failed, stage: stage_2} # 构造Prompt只让LLM分析指定文档 prompt f请从以下银行流水文本中提取月均收入单位元和近6个月最大单笔支出单位元{input_data[bank_statement_text]} schema { type: object, properties: { monthly_income: {type: number}, max_single_expense: {type: number} }, required: [monthly_income, max_single_expense] } try: result generate_structured_output(prompt, schema) self.state.update(result) return {status: PASSED, stage: stage_2, extracted_data: result} except Exception as e: return {status: FAILED, reason: str(e), stage: stage_2} def _stage_3_final_decision(self, input_data: dict) - dict: Stage 3: Final decision with hybrid model if not self.state.get(eligibility_passed) or not self.state.get(monthly_income): return {status: SKIPPED, reason: Prerequisites not met, stage: stage_3} # 准备特征向量 features [ self.state[monthly_income], self.state[max_single_expense], input_data.get(loan_amount, 0), get_credit_score(input_data[id_number]) # 调用征信API ] # 调用XGBoost模型 prediction xgb_model.predict([features])[0] probability xgb_model.predict_proba([features])[0][1] # 双模验证XGBoost输出 LLM对决策的解释 llm_explanation self._generate_explanation(input_data, features, prediction) # 综合判断 if probability 0.85 and llm_explanation[confidence_score] 0.8: decision APPROVED elif probability 0.15: decision REJECTED else: decision PENDING_HUMAN_REVIEW return { status: PASSED, stage: stage_3, decision: decision, probability: probability, llm_explanation: llm_explanation } def run(self, input_data: dict) - dict: 执行整个校验环 full_result {stages: [], final_decision: None, audit_trace: []} for stage_name, stage_func in self.stages: try: stage_result stage_func(input_data) full_result[stages].append(stage_result) # 记录审计轨迹 audit_entry { stage: stage_name, result: stage_result, timestamp: time.time(), state_snapshot: self.state.copy() } full_result[audit_trace].append(audit_entry) # 如果当前阶段失败且不可降级提前退出 if stage_result[status] REJECTED: full_result[final_decision] REJECTED break except Exception as e: error_result {status: ERROR, reason: str(e), stage: stage_name} full_result[stages].append(error_result) full_result[audit_trace].append({stage: stage_name, error: str(e)}) break # 设置最终决策 if not full_result[final_decision]: last_stage full_result[stages][-1] full_result[final_decision] last_stage.get(decision, UNKNOWN) return full_result # 使用示例 loop StagedValidationLoop() input_data { id_number: 11010119900307299X, bank_statement_text: 2023-01-01 收入 15000.00 ... 2023-06-30 支出 8500.00 } result loop.run(input_data) print(json.dumps(result, indent2, ensure_asciiFalse))实操心得阶段划分要符合业务直觉Bank of America的三个阶段对应的是信贷审批的真实业务流程先看人再看材料最后综合决策而不是技术便利性。每个阶段必须有明确的“失败即退出”条件Stage 1的“身份证无效”是硬性退出Stage 2的“LLM解析失败”可以降级为人工录入Stage 3的“概率临界”则必须人工复核。这种差异化的失败策略是效率与安全的平衡点。审计轨迹audit_trace是生命线它记录了每个阶段的输入、输出、状态快照和时间戳。当一个贷款被拒时风控部门能精确看到是Stage 1的黑名单命中还是Stage 3的模型概率不足这决定了申诉路径。状态state管理要谨慎只传递必要字段避免“状态污染”。UiPath建议用state[stage_2_income]而非state[income]以明确来源。3.4 模式4外部知识锚定External Knowledge Anchoring核心思想禁止Agent“凭空想象”所有关键判断、定义、计算都必须锚定到一个权威、可验证、版本化的外部知识源Knowledge Base, KB。LLM的角色仅限于“检索-理解-引用”而非“创造”。为什么必须用LLM的幻觉Hallucination在专业领域是灾难性的。在金融场景“美联储基准利率”是一个精确到小数点后两位的法定数值LLM如果“回忆”错了可能导致整个定价模型失效。锚定外部KB相当于给LLM装上了“事实核查器”。Coinbase实践细节Coinbase的合规Agent其所有监管条款引用都指向一个内部托管的、Git版本化的Markdown知识库。每条条款都有唯一ID如KB-2024-001Agent的输出中必须包含source_reference: KB-2024-001。当KB更新时所有引用该ID的Agent输出都会被自动标记为“待验证”触发回归测试。实操代码片段KB-aware RAG流程import markdown from bs4 import BeautifulSoup class KnowledgeAnchoredRAG: def __init__(self, kb_repo_path: str): self.kb_repo_path kb_repo_path self.kb_index self._build_kb_index() # 构建KB索引 def _build_kb_index(self) - dict: 构建KB索引{kb_id: {title: ..., content: ..., version: v1.2}} index {} for md_file in Path(self.kb_repo_path).rglob(*.md): with open(md_file, r, encodingutf-8) as f: content f.read() # 解析Markdown头部的YAML元数据 if content.startswith(---): meta_end content.find(---, 3) if meta_end 0: meta_yaml content[3:meta_end].strip() try: meta yaml.safe_load(meta_yaml) kb_id meta.get(id) if kb_id: # 提取纯文本内容去除Markdown格式 html markdown.markdown(content[meta_end3:]) soup BeautifulSoup(html, html.parser) plain_text soup.get_text() index[kb_id] { title: meta.get(title, ), content: plain_text, version: meta.get(version, v1.0), file_path: str(md_file) } except: pass return index def retrieve_relevant_kb(self, query: str, top_k: int 3) - list: 基于语义相似度检索相关KB条目 # 这里用伪代码实际可用Sentence-BERT或ColBERT # 为简洁我们用简单的TF-IDF 余弦相似度 from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity # 构建KB文本库 kb_texts [kb[content] for kb in self.kb_index.values()] vectorizer TfidfVectorizer(max_features5000, stop_wordsenglish) kb_vectors vectorizer.fit_transform(kb_texts) # 查询向量化 query_vector vectorizer.transform([query]) # 计算相似度 similarities cosine_similarity(query_vector, kb_vectors).flatten() top_indices similarities.argsort()[-top_k:][::-1] # 返回KB ID列表 kb_ids list(self.kb_index.keys()) return [kb_ids[i] for i in top_indices if similarities[i] 0.1] def generate_with_anchoring(self, user_query: str, context_kb_ids: list None) - dict: 生成带知识锚定的响应 返回: {answer: ..., citations: [{kb_id: KB-2024-001, snippet: ...}]} if not context_kb_ids: context_kb_ids self.retrieve_relevant_kb(user_query) # 构建上下文将KB内容拼接为Context context_parts [] citations [] for kb_id in context_kb_ids[:2]: # 最多引用2个KB kb_info self.kb_index.get(kb_id) if kb_info: # 截取相关片段简化版取前200字符 snippet kb_info[content][:200] ... context_parts.append(f【知识源 {kb_id}】{kb_info[title]}\n{snippet}) citations.append({ kb_id: kb_id, title: kb_info[title], version: kb_info[version], snippet: snippet }) context \n\n.join(context_parts) # 构造Prompt强制引用 prompt f你是一个严谨的合规助手。请基于以下提供的【知识源】回答问题所有结论必须有据可依。 【知识源】 {context} 【用户问题】 {user_query} 【要求】 1. 回答必须简洁、准确直接回应问题。 2. 如果答案在【知识源】中明确给出请直接引用不要添加推测。 3. 在回答末尾用JSON格式列出所有引用的知识源ID例如{{citations: [KB-2024-001, KB-2024-002]}} 4. 如果【知识源】中没有相关信息请回答依据当前知识库无法确定。请咨询合规部门。 请开始回答 # 调用LLM使用模式1确保JSON输出 schema { type: object, properties: { answer: {type:
生产级AI Agent的8种工程化模式：从银行到加密巨头的落地实践
1. 这不是概念演示是银行、加密巨头和RPA龙头每天真正在跑的AI代理模式“Production-Ready AI Agents”这个短语最近被讲得太多多到让人怀疑它是不是又一个PPT里的热词。但如果你真去看过Bank of America的内部技术简报、Coinbase的工程博客或者UiPath去年发布的客户案例白皮书你会发现一件事他们没在聊“未来可能怎么做”而是在说“上个月我们用这个Agent处理了237万笔反欺诈工单”“上周这个Agent自动完成了89%的客户开户KYC初审”“过去90天这个Agent为全球27个区域的财务团队节省了11,400小时人工核对时间”。这不是实验室Demo是嵌在核心业务流里、扛着SLA服务等级协议、要过审计、能接真实数据库、会写日志、能告警、出问题能回滚的AI代理——也就是真正意义上的Production-Ready。我带团队落地过6个跨行业的AI Agent项目从保险理赔到供应链调度最深的体会是90%的失败不是因为模型不够强而是因为从第一天起就把它当成了“调用一次大模型API的小脚本”而不是一个需要工程化设计、可观测、可运维、可审计的软件系统。Bank of America不用LangChain写信贷审批AgentCoinbase不拿Llama-3直接连交易所API做风控决策UiPath更不会让一个未经沙箱隔离的Agent直接修改ERP主数据——他们用的是一套经过千锤百炼的模式语言Pattern Language就像建筑行业有“承重墙剪力墙框架结构”的组合逻辑AI Agent的生产级落地也有它自己的、可复用、可验证、可审计的8种基础模式。这8种模式不是理论推演出来的而是从三家机构公开披露的技术文档、专利申请、GitHub开源片段、技术大会演讲逐行抠出来的。比如Bank of America在2023年Q3财报附录里提到的“multi-stage validation loop for loan underwriting”对应的是本文的模式3分阶段校验环Staged Validation LoopCoinbase工程师在PyCon演讲中展示的“real-time anomaly triage with fallback to human-in-the-loop”就是模式6异常熔断人工兜底Anomaly Circuit Breaker Human FallbackUiPath在2024年产品路线图里强调的“agent-to-agent handoff with state persistence”则精准匹配模式7状态持久化代理交接Stateful Agent Handoff。这些不是命名游戏每个模式背后都绑定了明确的触发条件、数据契约、错误边界、监控指标和回滚路径。接下来我会把这8个模式掰开揉碎告诉你它们在真实系统里长什么样、为什么必须这么设计、踩过哪些坑、怎么抄作业。2. 为什么这8种模式能活下来——从“能跑通”到“敢上线”的底层逻辑2.1 模式存在的根本动因对抗AI不确定性三座大山所有想把AI Agent推进生产环境的团队最终都会撞上三堵墙。这8种模式本质上就是针对这三堵墙的工程化应答方案。第一堵墙叫输出不可控性Output Unpredictability。大模型的输出不是确定性函数同一个Prompt输入微小扰动比如日期格式多一个空格结果可能从“批准贷款”变成“拒绝并标记高风险”。Bank of America的信贷审批Agent如果只依赖单次LLM调用哪怕准确率标称95%剩下5%的误判也意味着每年数千万美元的坏账或合规处罚。所以他们的方案是绝不让LLM直接做终局决策。模式1结构化输出约束强制LLM只生成JSON Schema定义的字段模式2双模验证器让规则引擎和LLM各自打分只有两者置信度都超阈值才放行模式4外部知识锚定要求所有判断必须引用KB中编号为KB-2023-087的监管条款原文。这不是限制AI而是给它画出安全运行的轨道。第二堵墙是上下文脆弱性Context Fragility。Agent在处理一个复杂任务时往往要跨多个步骤、调用多个工具、读取多份文档。但LLM的上下文窗口有限且长文本推理容易“忘记”前面的关键约束。UiPath的财务对账Agent曾出现过这样的故障第一步从邮件提取发票号第二步查ERP获取金额第三步比对银行流水——但在第三步时模型“忘了”第一步提取的发票号转而用当前日期生成了一个假号码去查询导致对账失败并触发错误付款。他们的解法是模式5显式状态快照即每完成一步就把关键中间状态如invoice_id: INV-2024-7891, erp_amount: 14280.50序列化成带版本号的键值对存入Redis后续步骤必须显式读取该键而非依赖上下文记忆。这看起来笨重但换来的是100%的状态可追溯。第三堵墙是系统耦合风险System Coupling Risk。很多团队一上来就想让Agent“全知全能”查数据库、调API、发邮件、改文件……结果一个下游服务抖动整个Agent链路雪崩。Coinbase的交易风控Agent曾因此在一次AWS us-east-1区域故障中误将正常交易标记为洗钱暂停了37分钟的现货交易。他们后来重构为模式8松耦合工具编排即Agent本身只做决策所有工具调用由独立的Tool Orchestrator服务执行该服务自带重试策略、熔断阈值、降级开关比如当交易所API超时自动切到本地缓存的行情快照。Agent与工具之间只传递标准化的Input/Output Schema不共享任何连接、线程或内存。提示这三堵墙不是“等模型变强就能消失”的问题而是AI作为概率性组件嵌入确定性系统时必然存在的工程矛盾。回避它等于把生产环境当沙盒正视它才能设计出真正可靠的模式。2.2 模式选择的黄金法则按业务SLA倒推技术选型选哪种模式从来不是看哪个“听起来酷”而是看你的业务场景对哪项指标最敏感。我整理了一张决策表这是我和Bank of America架构师一起喝咖啡时画在餐巾纸上的业务场景特征最敏感SLA指标推荐主导模式关键设计要点Bank of America实例高合规要求如反洗钱、信贷审批零误判容忍False Positive Rate 0.001%模式3分阶段校验环每阶段设独立阈值任一阶段失败即终止全程留痕供审计KYC初筛先规则引擎过滤剔除92%明显不合格再LLM分析模糊材料最后人工复核剩余8%高实时性要求如交易风控、IoT告警端到端延迟 800msP95模式6异常熔断人工兜底熔断阈值设为响应时间300ms或置信度0.85熔断后自动转人工队列并推送摘要实时交易监控当检测到异常模式若LLM分析耗时超阈值立即触发人工审核通道同时返回“待确认”状态长流程协作如客户开户、保险理赔流程成功率 99.5%且失败可精确归因模式7状态持久化代理交接每个Agent交接点生成唯一trace_id状态存入带TTL的Key-Value存储失败时可从任意checkpoint重放开户流程身份核验Agent → 资料录入Agent → 合规审查Agent每步状态存Redis超时自动告警这张表的核心逻辑是用确定性机制规则、阈值、状态存储去包裹不确定性组件LLM。比如当业务要求“零误判”时你不能寄希望于LLM自己变得无比准确而应该设计一个环路让LLM只负责它最擅长的部分理解模糊文本把最危险的终局决策权交给确定性更强的规则引擎或人工。这就像汽车的安全气囊——不是指望司机永远不犯错而是为错误准备确定性的缓冲。2.3 模式不是孤立的而是可组合的“乐高积木”现实中没有哪个Agent只用一种模式。真正的生产系统是多个模式的嵌套与组合。以Coinbase的“新用户资产安全初始化”Agent为例它完整串联了5种模式启动层模式1结构化输出约束——用户注册后Agent首先生成一个严格Schema的security_setup_plan.json包含required_steps: [2fa_enrollment, whitelist_address_addition, withdrawal_limit_set]等字段杜绝自由发挥执行层模式8松耦合工具编排——每个required_step被发送给独立的Tool Orchestrator由其调用2FA服务、地址白名单API、风控限额服务Agent只接收标准化结果校验层模式2双模验证器——当whitelist_address_addition返回成功Agent不直接信任而是调用链上扫描工具验证该地址是否真在白名单合约中双源一致才标记完成容错层模式6异常熔断人工兜底——若链上扫描超时5s立即熔断将任务推入人工审核队列并附带原始交易哈希和超时日志状态层模式7状态持久化代理交接——整个流程的plan_id、各步骤status、last_updated_timestamp存入DynamoDB供客服系统实时查询进度。这种组合不是随意堆砌而是遵循一个清晰的分层原则越靠近用户交互层越强调结构化与可控越靠近执行层越强调解耦与韧性越靠近决策层越强调验证与可溯。理解这一点你才能避免“为了用模式而用模式”的陷阱。3. 八大生产级AI Agent模式详解从原理、结构到实操代码片段3.1 模式1结构化输出约束Structured Output Constraint核心思想不让LLM“自由创作”而是强制它在一个预定义的、机器可解析的JSON Schema内填空。这是所有生产级Agent的起点也是最简单却最有效的防线。为什么必须用自由文本输出无法被程序直接消费。想象一下你的Agent返回“好的我已经把发票金额设为一万四千二百八十块五毛。”——下游系统怎么知道这个数字是14280.50还是1428050而如果它返回{invoice_amount: 14280.50, currency: USD}一切就变得确定、可校验、可审计。Bank of America实践细节他们在信贷审批Agent中使用OpenAI的response_format{type: json_object}参数并配合一个自研的Schema Validator。这个Validator不仅检查JSON语法还做业务级校验比如loan_amount必须是正数且小于max_approved_amount从规则引擎获取的动态值credit_score必须在300-850区间。任何校验失败Agent立即返回错误码ERR_SCHEMA_VIOLATION而非尝试“修复”输出。实操代码片段Python OpenAI SDKfrom openai import OpenAI import json import re client OpenAI() def generate_structured_output(prompt: str, schema: dict) - dict: schema 示例: { type: object, properties: { decision: {type: string, enum: [APPROVE, REJECT, PENDING]}, confidence_score: {type: number, minimum: 0.0, maximum: 1.0}, reasoning_summary: {type: string, maxLength: 500} }, required: [decision, confidence_score] } try: response client.chat.completions.create( modelgpt-4-turbo, messages[{role: user, content: prompt}], response_format{type: json_object}, temperature0.1, # 降低随机性 max_tokens500 ) # 1. 基础JSON解析 raw_json json.loads(response.choices[0].message.content) # 2. 业务Schema校验关键 for required_key in schema.get(required, []): if required_key not in raw_json: raise ValueError(fMissing required field: {required_key}) # 3. 数值范围校验 if confidence_score in raw_json: score raw_json[confidence_score] if not (0.0 score 1.0): raise ValueError(fconfidence_score out of range: {score}) return raw_json except json.JSONDecodeError as e: # 记录原始LLM输出用于调试 log_error(fJSON Parse Error: {response.choices[0].message.content[:200]}) raise except ValueError as e: log_error(fSchema Validation Error: {e}) raise # 使用示例 prompt 基于以下客户信息信用分720年收入$120,000负债比35%请做出贷款审批决策。 schema { type: object, properties: { decision: {type: string, enum: [APPROVE, REJECT, PENDING]}, confidence_score: {type: number, minimum: 0.0, maximum: 1.0}, reasoning_summary: {type: string, maxLength: 500} }, required: [decision, confidence_score] } result generate_structured_output(prompt, schema) print(result) # 输出: {decision: APPROVE, confidence_score: 0.92, reasoning_summary: 信用分和收入均高于阈值...}实操心得不要省略业务级校验OpenAI的response_format只保证JSON语法正确不保证业务逻辑。必须自己写校验逻辑这是生产环境的底线。温度值temperature设为0.1或更低生产环境不是创意写作确定性优先于多样性。为每个Schema定义明确的required字段避免LLM“偷懒”省略关键信息。记录原始LLM输出当校验失败时原始输出是调试的唯一线索务必存入日志。3.2 模式2双模验证器Dual-Mode Validator核心思想让LLM和一个确定性规则引擎Rule Engine对同一输入独立打分只有两者结论一致且置信度都达标时才采纳结果。这是对抗LLM“一本正经胡说八道”的终极保险。为什么必须用LLM可能给出逻辑完美但事实错误的答案。例如在保险理赔中LLM可能根据病历描述“剧烈腹痛、发热”推理出“急性阑尾炎”并给出高置信度0.95——但它忽略了病历末尾手写的“已排除阑尾炎确诊为胃肠炎”。规则引擎则会机械地匹配关键词“已排除阑尾炎”直接返回0分。双模验证能捕获这种致命偏差。UiPath实践细节UiPath在其RPAAI平台中将此模式封装为ConfidenceGate组件。它要求规则引擎输出一个rule_score0-100LLM输出一个llm_score0-1两者加权平均权重可配置需大于阈值如0.82才通过。更重要的是它强制记录两者的原始输出供后续审计。在一次医疗报销审核中正是通过对比发现LLM的推理链遗漏了关键否定词从而推动了Prompt优化。实操代码片段模拟双模验证流程# 规则引擎简化版实际可能是Drools或自研DSL def rule_engine_eval(input_data: dict) - float: 输入: {symptoms: [腹痛, 发热], diagnosis_notes: 已排除阑尾炎确诊为胃肠炎} 输出: 规则得分 (0-100) notes input_data.get(diagnosis_notes, ) # 硬编码规则出现已排除且后面跟着疾病名则得0分 if 已排除 in notes and any(disease in notes for disease in [阑尾炎, appendicitis]): return 0.0 # 其他规则... return 85.0 # LLM评估使用上一节的structured_output函数 def llm_eval(input_data: dict) - dict: prompt f患者症状{input_data[symptoms]}。诊断备注{input_data[diagnosis_notes]}。请输出JSON{{diagnosis: ..., confidence_score: 0.x}} schema { type: object, properties: { diagnosis: {type: string}, confidence_score: {type: number, minimum: 0.0, maximum: 1.0} }, required: [diagnosis, confidence_score] } return generate_structured_output(prompt, schema) def dual_mode_validate(input_data: dict, rule_weight: float 0.4, llm_weight: float 0.6, threshold: float 0.82) - dict: 双模验证主函数 返回: {valid: bool, final_score: float, rule_score: float, llm_score: float, llm_output: dict} rule_score rule_engine_eval(input_data) / 100.0 # 归一化到0-1 llm_output llm_eval(input_data) llm_score llm_output[confidence_score] final_score rule_weight * rule_score llm_weight * llm_score is_valid (final_score threshold) and (rule_score 0.0) # 规则引擎得分为0是硬性拒绝 # 记录详细日志生产环境必需 log_audit({ input_data: input_data, rule_score: rule_score, llm_score: llm_score, final_score: final_score, is_valid: is_valid, llm_output: llm_output, timestamp: time.time() }) return { valid: is_valid, final_score: final_score, rule_score: rule_score, llm_score: llm_score, llm_output: llm_output } # 使用示例 input_data { symptoms: [腹痛, 发热], diagnosis_notes: 已排除阑尾炎确诊为胃肠炎 } result dual_mode_validate(input_data) print(result) # 输出: {valid: False, final_score: 0.34, rule_score: 0.0, llm_score: 0.95, ...} # 尽管LLM信心很高但规则引擎一票否决实操心得规则引擎不是摆设它的规则必须来自领域专家且定期更新。UiPath要求所有规则变更必须附带业务负责人签字的变更说明。权重不是拍脑袋Bank of America在反洗钱场景中将规则引擎权重设为0.7因为监管条款是刚性的而在客户服务场景中LLM权重提至0.8因为用户体验更看重自然度。“规则得分为0”是硬性熔断点这代表存在明确的业务禁忌如“已排除某病”此时无论LLM多自信都必须拒绝。审计日志必须包含原始输入和两个模型的原始输出这是应对监管检查的唯一证据。3.3 模式3分阶段校验环Staged Validation Loop核心思想将一个复杂的决策过程拆解为多个逻辑上递进、技术上隔离的阶段每个阶段有自己的输入、输出、校验规则和失败处理策略。失败不直接终止而是降级到下一阶段或人工。为什么必须用单点校验太脆弱。比如信贷审批如果只在最后一步校验“总负债率50%”那么前面所有关于收入、资产的分析都可能白费。分阶段校验则像工厂流水线每个工位都有质检员问题在源头就被拦截。Bank of America实践细节他们的贷款审批Agent采用经典的三阶段环Stage 1准入过滤纯规则引擎。检查身份证有效性、黑名单、基础资质如年龄18。99%的申请在此阶段被快速拒绝不消耗LLM资源。Stage 2材料分析LLM主导。分析上传的收入证明、银行流水PDF提取关键字段。输出必须通过模式1的结构化约束。Stage 3综合决策混合模式。将Stage 1的规则结果、Stage 2的LLM提取数据、以及实时征信API返回的数据输入一个轻量级XGBoost模型生成最终评分。此阶段也应用模式2的双模验证。实操代码片段阶段化流程控制器class StagedValidationLoop: def __init__(self): self.stages [ (stage_1_eligibility, self._stage_1_eligibility), (stage_2_document_analysis, self._stage_2_document_analysis), (stage_3_final_decision, self._stage_3_final_decision) ] self.state {} # 共享状态字典 def _stage_1_eligibility(self, input_data: dict) - dict: Stage 1: Rule-based eligibility check # 检查身份证号格式 id_number input_data.get(id_number, ) if not re.match(r^\d{17}[\dXx]$, id_number): return {status: REJECTED, reason: Invalid ID format, stage: stage_1} # 检查年龄 birth_year int(id_number[6:10]) age 2024 - birth_year if age 18 or age 70: return {status: REJECTED, reason: Age out of range, stage: stage_1} # 检查黑名单 if is_in_blacklist(id_number): return {status: REJECTED, reason: In blacklist, stage: stage_1} # 通过进入下一阶段 self.state[eligibility_passed] True return {status: PASSED, stage: stage_1} def _stage_2_document_analysis(self, input_data: dict) - dict: Stage 2: LLM-based document analysis if not self.state.get(eligibility_passed): return {status: SKIPPED, reason: Stage 1 failed, stage: stage_2} # 构造Prompt只让LLM分析指定文档 prompt f请从以下银行流水文本中提取月均收入单位元和近6个月最大单笔支出单位元{input_data[bank_statement_text]} schema { type: object, properties: { monthly_income: {type: number}, max_single_expense: {type: number} }, required: [monthly_income, max_single_expense] } try: result generate_structured_output(prompt, schema) self.state.update(result) return {status: PASSED, stage: stage_2, extracted_data: result} except Exception as e: return {status: FAILED, reason: str(e), stage: stage_2} def _stage_3_final_decision(self, input_data: dict) - dict: Stage 3: Final decision with hybrid model if not self.state.get(eligibility_passed) or not self.state.get(monthly_income): return {status: SKIPPED, reason: Prerequisites not met, stage: stage_3} # 准备特征向量 features [ self.state[monthly_income], self.state[max_single_expense], input_data.get(loan_amount, 0), get_credit_score(input_data[id_number]) # 调用征信API ] # 调用XGBoost模型 prediction xgb_model.predict([features])[0] probability xgb_model.predict_proba([features])[0][1] # 双模验证XGBoost输出 LLM对决策的解释 llm_explanation self._generate_explanation(input_data, features, prediction) # 综合判断 if probability 0.85 and llm_explanation[confidence_score] 0.8: decision APPROVED elif probability 0.15: decision REJECTED else: decision PENDING_HUMAN_REVIEW return { status: PASSED, stage: stage_3, decision: decision, probability: probability, llm_explanation: llm_explanation } def run(self, input_data: dict) - dict: 执行整个校验环 full_result {stages: [], final_decision: None, audit_trace: []} for stage_name, stage_func in self.stages: try: stage_result stage_func(input_data) full_result[stages].append(stage_result) # 记录审计轨迹 audit_entry { stage: stage_name, result: stage_result, timestamp: time.time(), state_snapshot: self.state.copy() } full_result[audit_trace].append(audit_entry) # 如果当前阶段失败且不可降级提前退出 if stage_result[status] REJECTED: full_result[final_decision] REJECTED break except Exception as e: error_result {status: ERROR, reason: str(e), stage: stage_name} full_result[stages].append(error_result) full_result[audit_trace].append({stage: stage_name, error: str(e)}) break # 设置最终决策 if not full_result[final_decision]: last_stage full_result[stages][-1] full_result[final_decision] last_stage.get(decision, UNKNOWN) return full_result # 使用示例 loop StagedValidationLoop() input_data { id_number: 11010119900307299X, bank_statement_text: 2023-01-01 收入 15000.00 ... 2023-06-30 支出 8500.00 } result loop.run(input_data) print(json.dumps(result, indent2, ensure_asciiFalse))实操心得阶段划分要符合业务直觉Bank of America的三个阶段对应的是信贷审批的真实业务流程先看人再看材料最后综合决策而不是技术便利性。每个阶段必须有明确的“失败即退出”条件Stage 1的“身份证无效”是硬性退出Stage 2的“LLM解析失败”可以降级为人工录入Stage 3的“概率临界”则必须人工复核。这种差异化的失败策略是效率与安全的平衡点。审计轨迹audit_trace是生命线它记录了每个阶段的输入、输出、状态快照和时间戳。当一个贷款被拒时风控部门能精确看到是Stage 1的黑名单命中还是Stage 3的模型概率不足这决定了申诉路径。状态state管理要谨慎只传递必要字段避免“状态污染”。UiPath建议用state[stage_2_income]而非state[income]以明确来源。3.4 模式4外部知识锚定External Knowledge Anchoring核心思想禁止Agent“凭空想象”所有关键判断、定义、计算都必须锚定到一个权威、可验证、版本化的外部知识源Knowledge Base, KB。LLM的角色仅限于“检索-理解-引用”而非“创造”。为什么必须用LLM的幻觉Hallucination在专业领域是灾难性的。在金融场景“美联储基准利率”是一个精确到小数点后两位的法定数值LLM如果“回忆”错了可能导致整个定价模型失效。锚定外部KB相当于给LLM装上了“事实核查器”。Coinbase实践细节Coinbase的合规Agent其所有监管条款引用都指向一个内部托管的、Git版本化的Markdown知识库。每条条款都有唯一ID如KB-2024-001Agent的输出中必须包含source_reference: KB-2024-001。当KB更新时所有引用该ID的Agent输出都会被自动标记为“待验证”触发回归测试。实操代码片段KB-aware RAG流程import markdown from bs4 import BeautifulSoup class KnowledgeAnchoredRAG: def __init__(self, kb_repo_path: str): self.kb_repo_path kb_repo_path self.kb_index self._build_kb_index() # 构建KB索引 def _build_kb_index(self) - dict: 构建KB索引{kb_id: {title: ..., content: ..., version: v1.2}} index {} for md_file in Path(self.kb_repo_path).rglob(*.md): with open(md_file, r, encodingutf-8) as f: content f.read() # 解析Markdown头部的YAML元数据 if content.startswith(---): meta_end content.find(---, 3) if meta_end 0: meta_yaml content[3:meta_end].strip() try: meta yaml.safe_load(meta_yaml) kb_id meta.get(id) if kb_id: # 提取纯文本内容去除Markdown格式 html markdown.markdown(content[meta_end3:]) soup BeautifulSoup(html, html.parser) plain_text soup.get_text() index[kb_id] { title: meta.get(title, ), content: plain_text, version: meta.get(version, v1.0), file_path: str(md_file) } except: pass return index def retrieve_relevant_kb(self, query: str, top_k: int 3) - list: 基于语义相似度检索相关KB条目 # 这里用伪代码实际可用Sentence-BERT或ColBERT # 为简洁我们用简单的TF-IDF 余弦相似度 from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity # 构建KB文本库 kb_texts [kb[content] for kb in self.kb_index.values()] vectorizer TfidfVectorizer(max_features5000, stop_wordsenglish) kb_vectors vectorizer.fit_transform(kb_texts) # 查询向量化 query_vector vectorizer.transform([query]) # 计算相似度 similarities cosine_similarity(query_vector, kb_vectors).flatten() top_indices similarities.argsort()[-top_k:][::-1] # 返回KB ID列表 kb_ids list(self.kb_index.keys()) return [kb_ids[i] for i in top_indices if similarities[i] 0.1] def generate_with_anchoring(self, user_query: str, context_kb_ids: list None) - dict: 生成带知识锚定的响应 返回: {answer: ..., citations: [{kb_id: KB-2024-001, snippet: ...}]} if not context_kb_ids: context_kb_ids self.retrieve_relevant_kb(user_query) # 构建上下文将KB内容拼接为Context context_parts [] citations [] for kb_id in context_kb_ids[:2]: # 最多引用2个KB kb_info self.kb_index.get(kb_id) if kb_info: # 截取相关片段简化版取前200字符 snippet kb_info[content][:200] ... context_parts.append(f【知识源 {kb_id}】{kb_info[title]}\n{snippet}) citations.append({ kb_id: kb_id, title: kb_info[title], version: kb_info[version], snippet: snippet }) context \n\n.join(context_parts) # 构造Prompt强制引用 prompt f你是一个严谨的合规助手。请基于以下提供的【知识源】回答问题所有结论必须有据可依。 【知识源】 {context} 【用户问题】 {user_query} 【要求】 1. 回答必须简洁、准确直接回应问题。 2. 如果答案在【知识源】中明确给出请直接引用不要添加推测。 3. 在回答末尾用JSON格式列出所有引用的知识源ID例如{{citations: [KB-2024-001, KB-2024-002]}} 4. 如果【知识源】中没有相关信息请回答依据当前知识库无法确定。请咨询合规部门。 请开始回答 # 调用LLM使用模式1确保JSON输出 schema { type: object, properties: { answer: {type:
