From 0dd362bfd2edce9046512e5eb0fd3f8fcb814bc6 Mon Sep 17 00:00:00 2001 From: zuowei1216 <12206728+zuowei1216@user.noreply.gitee.com> Date: Fri, 12 Jun 2026 13:50:12 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E6=94=B9=E7=94=A8=20pydantic-ai=20?= =?UTF-8?q?=E6=A1=86=E6=9E=B6=E8=B0=83=E7=94=A8=E8=B1=86=E5=8C=85=EF=BC=8C?= =?UTF-8?q?=E7=BB=93=E6=9E=84=E5=8C=96=E8=BE=93=E5=87=BA=20PlateOrder?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app.py | 96 +++++++++++++++++++++++++++++------------------- requirements.txt | 1 + 2 files changed, 60 insertions(+), 37 deletions(-) diff --git a/app.py b/app.py index 3f4a45d..695ee7f 100644 --- a/app.py +++ b/app.py @@ -1,6 +1,6 @@ """ 开版订单 AI 解析服务 -Flask + 豆包 API（/api/v3/responses 格式） +Flask + pydantic-ai + 豆包（方舟 OpenAI 兼容协议 /chat/completions） 功能： 1. 代理企微消息接口（解决 CORS） @@ -17,12 +17,19 @@ import json import logging import threading import time +from typing import Optional + import requests from datetime import datetime, timedelta from flask import Flask, request, jsonify from flask_cors import CORS from dotenv import load_dotenv +from pydantic import BaseModel, Field, field_validator +from pydantic_ai import Agent +from pydantic_ai.models.openai import OpenAIChatModel +from pydantic_ai.providers.openai import OpenAIProvider + load_dotenv() # ─── 日志 ───────────────────────────────────────────────────────────────────── @@ -36,15 +43,10 @@ logger = logging.getLogger(__name__) app = Flask(__name__) CORS(app, origins="*") -# ─── 豆包 AI 配置 ───────────────────────────────────────────────────────────── +# ─── 豆包 AI 配置（通过 pydantic-ai + OpenAI 兼容协议调用方舟）───────────────── AI_API_KEY = os.getenv("AI_API_KEY", "") AI_BASE_URL = os.getenv("AI_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3") AI_MODEL = os.getenv("AI_MODEL", "doubao-seed-2-0-mini-260428") -DOUBAO_URL = f"{AI_BASE_URL}/responses" -DOUBAO_HEADERS = { - "Authorization": f"Bearer {AI_API_KEY}", - "Content-Type": "application/json", -} # ─── 企微消息服务配置 ───────────────────────────────────────────────────────── WECHAT_API_BASE = "https://test.ruicaiyinhua.online" @@ -331,37 +333,57 @@ def build_system_prompt() -> str: 3. 一律返回 JSON 对象，不要任何其它文字""" +class PlateOrder(BaseModel): + """开版订单解析结果。所有字段均可为 null（无法判断时返回 None）。""" + customer_name: Optional[str] = Field(None, description="客户名称（自由文本，消息开头）") + plate_type: Optional[str] = Field(None, description="起版情况枚举") + urgency_level: Optional[str] = Field(None, description="紧急程度枚举") + production_method: Optional[str] = Field(None, description="做货方式枚举") + plate_method: Optional[str] = Field(None, description="开版方式枚举") + area: Optional[str] = Field(None, description="客户区域枚举") + fabric: Optional[str] = Field(None, description="面料名称（自由文本，保留克重与布类）") + width: Optional[str] = Field(None, description="面料幅宽枚举") + fabric_source: Optional[str] = Field(None, description="面料来源枚举") + drawing_rating: Optional[str] = Field(None, description="画图评级枚举") + color_matching_rating: Optional[str] = Field(None, description="调色评级枚举") + sample_rating: Optional[str] = Field(None, description="套样评级枚举") + difficulty_rating: Optional[str] = Field(None, description="难度评级枚举") + process_name: Optional[str] = Field(None, description="关联流程名称") + plate_notes: Optional[str] = Field(None, description="打版备注（自由文本，原文保留）") + style_name: Optional[str] = Field(None, description="款号名称") + original_design_code: Optional[str] = Field(None, description="原订单ID（5-7位纯数字）") + merchandiser_name: Optional[str] = Field(None, description="跟单员（取 sender 原值）") + + @field_validator("*", mode="before") + @classmethod + def _normalize_empty(cls, v): + """模型有时会填字符串 'null'/'none'/空串，统一归一为 None""" + if isinstance(v, str) and v.strip().lower() in ("", "null", "none", "n/a", "无"): + return None + return v + + +# pydantic-ai Agent：通过 OpenAI 兼容协议调用方舟（豆包） +_provider = OpenAIProvider(base_url=AI_BASE_URL, api_key=AI_API_KEY) +_model = OpenAIChatModel(AI_MODEL, provider=_provider) +plate_agent = Agent( + _model, + output_type=PlateOrder, + model_settings={"temperature": 0.1, "max_tokens": 4096}, + retries=2, +) + + +@plate_agent.system_prompt +def _dynamic_system_prompt() -> str: + """系统提示词依赖运行时加载的字典缓存，故每次调用动态生成""" + return build_system_prompt() + + def call_doubao(user_text: str) -> dict: - """调用豆包 /v3/responses，返回解析后的 dict""" - payload = { - "model": AI_MODEL, - "input": [ - {"role": "system", "content": [{"type": "input_text", "text": build_system_prompt()}]}, - {"role": "user", "content": [{"type": "input_text", "text": user_text}]}, - ], - "temperature": 0.1, - "max_output_tokens": 4096, - } - r = requests.post(DOUBAO_URL, headers=DOUBAO_HEADERS, json=payload, timeout=60) - r.raise_for_status() - data = r.json() - - raw_text = "" - for item in data.get("output", []): - if item.get("type") == "message": - for c in item.get("content", []): - if c.get("type") == "output_text": - raw_text = c.get("text", "").strip() - break - break - if not raw_text: - raise ValueError("豆包返回内容为空") - - start = raw_text.find("{") - end = raw_text.rfind("}") + 1 - if start == -1 or end == 0: - raise ValueError(f"响应中未找到 JSON：{raw_text}") - return json.loads(raw_text[start:end]) + """通过 pydantic-ai 调用豆包，返回结构化结果 dict""" + result = plate_agent.run_sync(user_text) + return result.output.model_dump() # ─── 路由 ───────────────────────────────────────────────────────────────────── diff --git a/requirements.txt b/requirements.txt index 3fe7025..cbde001 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ flask==3.0.3 flask-cors==4.0.1 requests==2.32.3 python-dotenv==1.0.1 +pydantic-ai-slim[openai]==1.107.0