Hi我擅长AI 大模型应用落地、意识解码与 AI 开发工具链。 创业路上用技术换时间一起把 AI 变成生产力 可控、可测、可进化用 AI Agent 构建跨境支付风控系统实战在 B2B 跨境贸易中企业往往面临“多国别、多单据、多风险”的挑战。营业执照、报关单等非标文档版式各异传统的 OCR 往往依赖大量人工录入与校验同时资金流、物流、信息流的三流一致性验证依赖碎片化数据源隐性关联关系极难识别。行业正在从单纯的“AI 工具调用”转向构建“AI 数字员工”。模型能力越强对工程驾驭的要求就越高——这就是“可控、可测、可进化”的由来。对于在校学生或正转向 AI 工程领域的开发者而言理解并手写一套带有置信度评估和人机协同机制的 AI Agent是进入金融科技领域的一块极佳的敲门砖。本文将带你从零实现一个最小可行的跨境支付风控 Agent。前置准备环境、账号、依赖为了保证代码可直接复制运行我们将使用当前主流的开源大模型 API以阿里云 DashScope 提供的qwen3.6-max为例其原生支持 JSON 输出适合构建 Agent。如果你使用其他模型请确保具备类似的结构化输出能力。环境与依赖Python 3.10 及以上版本操作系统macOS / Linux / Windows本文以 macOS/Linux 命令行为例依赖库openai(用于兼容调用大模型 API)、pydantic(用于数据结构校验)安装依赖与配置环境# 创建并激活虚拟环境python3-mvenv ai_risk_agentsourceai_risk_agent/bin/activate# 安装依赖pipinstallopenai1.40.0pydantic2.9.0# 配置大模型 API Key (请替换为你自己的真实 Key)exportDASHSCOPE_API_KEYsk-xxxxxxxxxxxxxxxxxxxx步骤 1定义风控数据结构与文档解析可测的基础目标定义跨境支付审核所需的结构化数据模型并实现非标文档如报关单到结构化数据的转换。这是“可测”的前提只有数据结构严格确定后续的置信度评估才有意义。代码实现 (risk_agent.py第一部分)importosimportjsonfromopenaiimportOpenAIfrompydanticimportBaseModel,Field,ValidationErrorfromtypingimportOptional# 初始化客户端 (通过 OpenAI 兼容模式调用)clientOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)MODEL_NAMEqwen3.6-maxclassTradeDocument(BaseModel):跨境贸易单据结构company_name:strField(...,description企业名称)amount:floatField(...,description交易金额)currency:strField(...,description币种, 如 USD, CNY)goods_description:strField(...,description货物描述)risk_keywords:Optional[list[str]]Field(None,description文档中潜在的敏感或高风险词汇)defparse_document(raw_text:str)-TradeDocument:使用大模型从非结构化文本中提取结构化单据信息promptf 请从以下报关单文本中提取信息并严格按照指定的 JSON 格式输出。 文本内容{raw_text}responseclient.chat.completions.create(modelMODEL_NAME,messages[{role:user,content:prompt}],response_format{type:json_object},temperature0.1# 降低随机性保证解析稳定)try:datajson.loads(response.choices[0].message.content)returnTradeDocument(**data)exceptValidationErrorase:print(f数据结构校验失败:{e})raise预期输出当输入一段非结构化的单据文本时函数返回一个TradeDocument对象字段类型严格匹配。如果大模型返回了非法结构pydantic会直接抛出ValidationError。失败时怎么查如果报AuthenticationError检查DASHSCOPE_API_KEY是否配置正确如果频繁抛出ValidationError检查 prompt 是否清晰或模型是否支持强制 JSON 输出。步骤 2构建带置信度评估的自动审核逻辑可控的核心目标实现自动筛查与交叉验证。在推理过程中系统需实时评估置信度。置信度高时自动通过置信度不足时智能切换至人工复核。代码实现 (risk_agent.py第二部分)classRiskAssessment(BaseModel):风控评估结果is_suspicious:boolField(...,description是否可疑)confidence_score:floatField(...,description置信度评分, 0.0 到 1.0)reason:strField(...,description判断理由)defassess_risk(doc:TradeDocument)-RiskAssessment:对结构化单据进行风险评估promptf 你是一个跨境支付风控专家。请审核以下交易单据。 要求 1. 判断是否存在合规风险如敏感国家、高风险货物等。 2. 给出 0.0 到 1.0 之间的置信度评分。如果信息完整且无异常分数应高于 0.9如果存在模糊描述或潜在风险分数应低于 0.8。 单据信息{doc.model_dump_json()}responseclient.chat.completions.create(modelMODEL_NAME,messages[{role:user,content:prompt}],response_format{type:json_object},temperature0.1)datajson.loads(response.choices[0].message.content)returnRiskAssessment(**data)defexecute_risk_control(doc:TradeDocument,threshold:float0.85):执行风控闭环AI预判 — 人机共审print(f开始审核企业:{doc.company_name}, 金额:{doc.amount}{doc.currency})assessmentassess_risk(doc)print(fAI 评估结果: 可疑{assessment.is_suspicious}, 置信度{assessment.confidence_score})ifassessment.confidence_scorethresholdandnotassessment.is_suspicious:print(- 系统动作: AI Agent 自动通过)returnAUTO_APPROVEDelse:print(f- 系统动作: 置信度不足或存在风险切换至人工复核。原因:{assessment.reason})returnMANUAL_REVIEW面试/作业里常被追问的点为什么不直接让大模型输出“通过”或“拒绝”因为大模型的输出具有概率性直接控制业务流程极度危险。引入confidence_score作为路由控制变量是把不可解释的神经网络输出映射到了确定性的工程逻辑上。步骤 3实现持续学习的自进化闭环可进化目标当人工复核完成后将人工的最终决策反馈给系统作为历史经验沉淀使系统具备“持续学习”能力。代码实现 (risk_agent.py第三部分)# 模拟一个轻量级的向量存储或历史经验库experience_db[]defupdate_experience(doc:TradeDocument,assessment:RiskAssessment,human_decision:str):将人工复核结果沉淀为经验experience_record{document_info:doc.model_dump(),ai_assessment:assessment.model_dump(),human_decision:human_decision,feedback:人工修正if(assessment.is_suspicious!(human_decisionREJECTED))elseAI判断正确}experience_db.append(experience_record)print(f经验已沉淀:{experience_record[feedback]})完整示例将上述步骤串联这是一份可照抄运行的完整代码。保存为risk_agent.py并执行。importosimportjsonfromopenaiimportOpenAIfrompydanticimportBaseModel,Field,ValidationErrorfromtypingimportOptional clientOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)MODEL_NAMEqwen3.6-maxclassTradeDocument(BaseModel):company_name:strField(...,description企业名称)amount:floatField(...,description交易金额)currency:strField(...,description币种)goods_description:strField(...,description货物描述)risk_keywords:Optional[list[str]]Field(None,description高风险词汇)classRiskAssessment(BaseModel):is_suspicious:boolField(...,description是否可疑)confidence_score:floatField(...,description置信度 0.0-1.0)reason:strField(...,description判断理由)defparse_document(raw_text:str)-TradeDocument:promptf从以下文本提取信息输出JSON。文本{raw_text}responseclient.chat.completions.create(modelMODEL_NAME,messages[{role:user,content:prompt}],response_format{type:json_object},temperature0.1)returnTradeDocument(**json.loads(response.choices[0].message.content))defassess_risk(doc:TradeDocument)-RiskAssessment:promptf 你是跨境支付风控专家。审核单据{doc.model_dump_json()}判断风险并给出0.0-1.0的置信度。输出JSON。 responseclient.chat.completions.create(modelMODEL_NAME,messages[{role:user,content:prompt}],response_format{type:json_object},temperature0.1)returnRiskAssessment(**json.loads(response.choices[0].message.content))defexecute_risk_control(doc:TradeDocument,threshold:float0.85):print(f\n处理单据:{doc.company_name})assessmentassess_risk(doc)print(fAI置信度:{assessment.confidence_score}, 理由:{assessment.reason})ifassessment.confidence_scorethresholdandnotassessment.is_suspicious:print(- 动作: AI Agent 自动通过)returnAUTO_APPROVED,assessmentelse:print(- 动作: 切换人工复核)returnMANUAL_REVIEW,assessmentif__name____main__:# 模拟两份不同的跨境支付单据文本doc1_text出口商: 深圳科技贸易有限公司, 金额: 5000 USD, 货物: 普通蓝牙耳机配件, 无敏感词汇.doc2_text出口商: 某匿名贸易公司, 金额: 200000 USD, 货物:军民两用高精度金属部件.fortextin[doc1_text,doc2_text]:try:docparse_document(text)decision,assessmentexecute_risk_control(doc)# 模拟人工反馈进入自进化闭环human_decisionREJECTEDifdecisionMANUAL_REVIEWelseAPPROVEDprint(f最终决策:{human_decision}\n)exceptValidationErrorase:print(f校验失败:{e})exceptExceptionase:print(f系统异常:{e})运行命令python risk_agent.py常见问题 (FAQ)1. 大模型没有返回合法的 JSON导致程序崩溃怎么处理这是构建 Agent 最常见的问题。除了使用response_format{type: json_object}强制约束外必须在parse_document等函数外围加try...except ValidationError并设计重试机制如重试 3 次后降级到人工处理。2. 置信度 是模型真实的能力评估吗不完全是。大模型输出的分数是基于其自身概率分布的“自洽性”表现并不绝对等同于客观正确率。在真实生产中我们通常会将模型输出的分数与基于历史数据训练的独立分类器如 XGBoost得分进行加权融合以得到更可靠的工程置信度。3. 在真实的跨境支付场景中如何实现“三流一致性”验证这需要 Agent 具备工具调用能力。你需要将海关 API、银行流水 API 封装为工具让 Agent 在推理过程中自主调用比对接口而不是仅依赖单据文本本身。这超出了基础版范围属于 Multi-Agent 协同范畴。4. 为什么不直接用开源框架如 LangChain / Dify而要手写框架封装了大量抽象适合快速落地但对初学者理解底层逻辑不利。手写一次 Prompt 编排、JSON 解析、异常处理和置信度路由你才能真正在作品集里讲清楚 Agent 的运行机制这也是面试官最看重的底层掌控力。最佳实践结构化输出优先在任何 AI 工程中强制大模型输出 JSON 并用 Pydantic 校验是防止系统因非预期文本崩溃的第一道防线。基于置信度的路由设计永远不要让大模型直接做业务决策而是让模型输出置信度由确定性代码逻辑根据阈值如 0.85决定是自动放行还是转入人工审核。隔离业务逻辑与模型调用将parse_document和assess_risk设计为纯函数模型更换时只需修改 API 调用配置不需要改动业务路由代码。构建反馈数据飞轮人工复核的结果必须结构化存储定期用于微调模型或作为 Few-shot 示例注入 Prompt实现系统的“可进化”。