👋 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# 安装依赖pipinstallopenai==1.40.0pydantic==2.9.0# 配置大模型 API Key (请替换为你自己的真实 Key)exportDASHSCOPE_API_KEY="sk-xxxxxxxxxxxxxxxxxxxx"步骤 1：定义风控数据结构与文档解析（可测的基础）
目标：定义跨境支付审核所需的结构化数据模型，并实现非标文档（如报关单）到结构化数据的转换。这是“可测”的前提：只有数据结构严格确定，后续的置信度评估才有意义。
代码实现 (risk_agent.py第一部分)：
importosimportjsonfromopenaiimportOpenAIfrompydanticimportBaseModel,Field,ValidationErrorfromtypingimportOptional# 初始化客户端 (通过 OpenAI 兼容模式调用)client=OpenAI(api_key=os.getenv("DASHSCOPE_API_KEY"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")MODEL_NAME="qwen3.6-max"classTradeDocument(BaseModel):"""跨境贸易单据结构"""company_name:str=Field(...,description="企业名称")amount:float=Field(...,description="交易金额")currency:str=Field(...,description="币种, 如 USD, CNY")goods_description:str=Field(...,description="货物描述")risk_keywords:Optional[list[str]]=Field(None,description="文档中潜在的敏感或高风险词汇")defparse_document(raw_text:str)->TradeDocument:"""使用大模型从非结构化文本中提取结构化单据信息"""prompt=f""" 请从以下报关单文本中提取信息，并严格按照指定的 JSON 格式输出。 文本内容：{raw_text}"""response=client.chat.completions.create(model=MODEL_NAME,messages=[{"role":"user","content":prompt}],response_format={"type":"json_object"},temperature=0.1# 降低随机性，保证解析稳定)try:data=json.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:bool=Field(...,description="是否可疑")confidence_score:float=Field(...,description="置信度评分, 0.0 到 1.0")reason:str=Field(...,description="判断理由")defassess_risk(doc:TradeDocument)->RiskAssessment:"""对结构化单据进行风险评估"""prompt=f""" 你是一个跨境支付风控专家。请审核以下交易单据。 要求： 1. 判断是否存在合规风险（如敏感国家、高风险货物等）。 2. 给出 0.0 到 1.0 之间的置信度评分。如果信息完整且无异常，分数应高于 0.9；如果存在模糊描述或潜在风险，分数应低于 0.8。 单据信息：{doc.model_dump_json()}"""response=client.chat.completions.create(model=MODEL_NAME,messages=[{"role":"user","content":prompt}],response_format={"type":"json_object"},temperature=0.1)data=json.loads(response.choices[0].message.content)returnRiskAssessment(**data)defexecute_risk_control(doc:TradeDocument,threshold:float=0.85):"""执行风控闭环：AI预判 — 人机共审"""print(f"开始审核企业:{doc.company_name}, 金额:{doc.amount}{doc.currency}")assessment=assess_risk(doc)print(f"AI 评估结果: 可疑={assessment.is_suspicious}, 置信度={assessment.confidence_score}")ifassessment.confidence_score>=thresholdandnotassessment.is_suspicious:print("-> 系统动作: AI Agent 自动通过")return"AUTO_APPROVED"else:print(f"-> 系统动作: 置信度不足或存在风险，切换至人工复核。原因:{assessment.reason}")return"MANUAL_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_decision=="REJECTED"))else"AI判断正确"}experience_db.append(experience_record)print(f"经验已沉淀:{experience_record['feedback']}")完整示例
将上述步骤串联，这是一份可照抄运行的完整代码。保存为risk_agent.py并执行。
importosimportjsonfromopenaiimportOpenAIfrompydanticimportBaseModel,Field,ValidationErrorfromtypingimportOptional client=OpenAI(api_key=os.getenv("DASHSCOPE_API_KEY"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")MODEL_NAME="qwen3.6-max"classTradeDocument(BaseModel):company_name:str=Field(...,description="企业名称")amount:float=Field(...,description="交易金额")currency:str=Field(...,description="币种")goods_description:str=Field(...,description="货物描述")risk_keywords:Optional[list[str]]=Field(None,description="高风险词汇")classRiskAssessment(BaseModel):is_suspicious:bool=Field(...,description="是否可疑")confidence_score:float=Field(...,description="置信度 0.0-1.0")reason:str=Field(...,description="判断理由")defparse_document(raw_text:str)->TradeDocument:prompt=f"从以下文本提取信息，输出JSON。文本：{raw_text}"response=client.chat.completions.create(model=MODEL_NAME,messages=[{"role":"user","content":prompt}],response_format={"type":"json_object"},temperature=0.1)returnTradeDocument(**json.loads(response.choices[0].message.content))defassess_risk(doc:TradeDocument)->RiskAssessment:prompt=f""" 你是跨境支付风控专家。审核单据：{doc.model_dump_json()}判断风险并给出0.0-1.0的置信度。输出JSON。 """response=client.chat.completions.create(model=MODEL_NAME,messages=[{"role":"user","content":prompt}],response_format={"type":"json_object"},temperature=0.1)returnRiskAssessment(**json.loads(response.choices[0].message.content))defexecute_risk_control(doc:TradeDocument,threshold:float=0.85):print(f"\n处理单据:{doc.company_name}")assessment=assess_risk(doc)print(f"AI置信度:{assessment.confidence_score}, 理由:{assessment.reason}")ifassessment.confidence_score>=thresholdandnotassessment.is_suspicious:print("-> 动作: AI Agent 自动通过")return"AUTO_APPROVED",assessmentelse:print("-> 动作: 切换人工复核")return"MANUAL_REVIEW",assessmentif__name__=="__main__":# 模拟两份不同的跨境支付单据文本doc1_text="出口商: 深圳科技贸易有限公司, 金额: 5000 USD, 货物: 普通蓝牙耳机配件, 无敏感词汇."doc2_text="出口商: 某匿名贸易公司, 金额: 200000 USD, 货物:军民两用高精度金属部件."fortextin[doc1_text,doc2_text]:try:doc=parse_document(text)decision,assessment=execute_risk_control(doc)# 模拟人工反馈进入自进化闭环human_decision="REJECTED"ifdecision=="MANUAL_REVIEW"else"APPROVED"print(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，实现系统的“可进化”。
