企业官网建设流程全解析

如何仿制一个网站,dw网页设计成品免费下载,重庆制作网站公司,功能网站建设从PDF到决策#xff1a;构建生产就绪的AI系统自动化票据处理流程 TL;DR — 你将获得的内容 一套可运行、面向生产的设计方案#xff0c;包含完整的代码片段#xff0c;用于构建一个AI代理。该代理能够处理发票PDF文件、运行OCR、使用大语言模型提取结构化字段、验证数据、应…从PDF到决策构建生产就绪的AI系统自动化票据处理流程TL;DR — 你将获得的内容一套可运行、面向生产的设计方案包含完整的代码片段用于构建一个AI代理。该代理能够处理发票PDF文件、运行OCR、使用大语言模型提取结构化字段、验证数据、应用确定性业务逻辑并存储可审计的结果。所有内容都经过精心编写工程师可以直接复制粘贴到代码仓库中并在本地运行。包含测试、数据库迁移、指标监控和操作控制。语气实用且有主见。本文旨在通过代码审查、合规性问题和凌晨三点的生产告警——而不是为了赢得奖项竞赛。AI获胜的关键不在于聪明而在于可靠。为什么你应该关注大多数生成式AI文章都止步于单一的大语言模型调用。在实践中真正的系统会在那些看似无聊但至关重要的基础设施方面出问题——例如数据摄取的边界情况、重试机制、可观测性缺口以及静默的验证错误。本文展示了如何构建一个AI系统——而不是一个演示——来替代人工发票处理流程同时保持可审计性、安全性和可维护性。仓库结构ai-agent-invoice/ ├─ docker-compose.yml ├─ Dockerfile ├─ requirements.txt ├─ alembic/ │ └─ versions/001_create_tables.py ├─ app/ │ ├─ main.py │ ├─ config.py │ ├─ services/ │ │ ├─ ingestion.py │ │ ├─ preprocess.py │ │ ├─ ocr.py │ │ ├─ extractor.py │ │ ├─ llm_adapters/ │ │ ├─ validator.py │ │ ├─ decision.py │ │ └─ storage.py │ ├─ models/invoice.py │ ├─ db/repo.py │ └─ utils/ │ ├─ logging.py │ ├─ metrics.py │ └─ retry.py └─ infra/prometheus.yml快速浏览清单架构先行Pydantic v2模型是唯一的真相来源。大语言模型作为提取器而非决策器。业务规则存在于代码中。使用Decimal处理金额使用dateutil处理日期。使用流式上传采用幂等性和内容哈希。持久化原始PDF、OCR输出和模型元数据以便审计。使用后台工作器开发环境BackgroundTasks生产环境Celery/RQ/Kafka。全面监控Prometheus指标、结构化JSON日志和追踪。如果在六个月后无法解释某个决策的原因那么这个系统就是不完整的。高层流程运行时顺序客户端上传PDF →/ingest端点流式传输文件并返回document_id后台工作器将PDF转换为图像进行去歪斜和清理对每个页面运行OCR本地Tesseract或第三方服务→ 页面级文本和置信度分数大语言模型使用严格的模式提取结构化字段 → JSON优先使用函数调用/结构化输出自由文本解析是最后的手段使用Pydantic进行规范化与验证Decimal用于金额ISO格式日期确定性决策引擎自动批准或将发票加入人工审核队列持久化结果process_results JSONB provider_meta并发出指标和结构化日志关键设计原则简洁清晰架构先行。架构是真相的来源。大语言模型提取到架构字段中Pydantic验证它们。关注点分离。在不改变业务逻辑的情况下更换OCR或大语言模型提供商。确定性决策。如果你需要确定性的结果请使用版本化的策略将其编码在代码中。审计一切。持久化原始PDF、OCR文本、大语言模型输出、模型和修订详情以及所有审核操作。默认安全失败。低置信度的数据路由到人工审核。没有静默的自动批准。幂等性。使用X-Idempotency-Key加上内容哈希来避免重复处理。核心文件这里只包含最核心的文件——那些审查者会首先阅读的文件。将它们放入仓库布局中所示的路径根据你的环境进行调整并根据需要进行修改。app/models/invoice.py — 架构Pydantic v2# app/models/invoice.pyfrompydanticimportBaseModel,FieldfromdecimalimportDecimalfromdatetimeimportdateclassInvoice(BaseModel):invoice_number:strinvoice_date:date vendor_name:strtotal_amount:DecimalField(...,gtDecimal(0))currency:str注意使用Decimal避免金额四舍五入错误。日期存储为ISO格式。app/config.py — 设置环境驱动# app/config.pyfrompydanticimportBaseSettingsclassSettings(BaseSettings):DATABASE_URL:strMISTRAL_API_KEY:str|NoneNoneMAX_UPLOAD_BYTES:int20*1024*1024# 20MB defaultAPPROVAL_THRESHOLD:float5000.0classConfig:env_file.envsettingsSettings()app/services/ingestion.py — 流式传输 幂等性 内容哈希# app/services/ingestion.pyimportos,uuid,hashlibfromfastapiimportUploadFilefromapp.db.repoimportsave_document,get_document_by_idempotencyfromapp.utils.retryimportretry_backofffromapp.configimportsettings UPLOAD_DIRos.getenv(UPLOAD_DIR,/data/documents)os.makedirs(UPLOAD_DIR,exist_okTrue)retry_backoff()asyncdefsave_uploaded_file(file:UploadFile,idempotency_key:str|NoneNone)-str:ifidempotency_key:existingget_document_by_idempotency(idempotency_key)ifexisting:returnexisting.document_id document_idstr(uuid.uuid4())pathos.path.join(UPLOAD_DIR,f{document_id}.pdf)size0withopen(path,wb)asout_f:whileTrue:chunkawaitfile.read(1024*1024)ifnotchunk:breaksizelen(chunk)ifsizesettings.MAX_UPLOAD_BYTES:out_f.close()os.remove(path)raiseValueError(File too large)out_f.write(chunk)sha256hashlib.sha256()withopen(path,rb)asf:forblockiniter(lambda:f.read(65536),b):sha256.update(block)content_hashsha256.hexdigest()save_document(document_iddocument_id,pathpath,idempotency_keyidempotency_key,content_hashcontent_hash)returndocument_id流式传输可防止内存不足问题内容哈希和幂等键可防止重复处理。我们在同一个发票以不同文件名上传两次并自动处理两次后艰难地学到了这一点。app/services/preprocess.py — PDF→图像 去歪斜# app/services/preprocess.pyfrompdf2imageimportconvert_from_path,convert_from_bytesimportcv2,numpyasnpimporttempfile,osdefpreprocess_pdf_to_images(pdf_path:str,dpi:int300)-list[str]:imagesconvert_from_path(pdf_path,dpidpi)out_paths[]fori,pil_imginenumerate(images):imgcv2.cvtColor(np.array(pil_img),cv2.COLOR_RGB2BGR)img_deskew_image_safe(img)tmp_pathos.path.join(tempfile.gettempdir(),f{os.path.basename(pdf_path)}_page_{i}.png)cv2.imwrite(tmp_path,img)out_paths.append(tmp_path)returnout_pathsdef_deskew_image_safe(img):try:graycv2.cvtColor(img,cv2.COLOR_BGR2GRAY)coordsnp.column_stack(np.where(gray0))ifcoords.size0:returnimg anglecv2.minAreaRect(coords)[-1]ifangle-45:angle-(90angle)else:angle-angle(h,w)img.shape[:2]Mcv2.getRotationMatrix2D((w//2,h//2),angle,1.0)rotatedcv2.warpAffine(img,M,(w,h),flagscv2.INTER_CUBIC,borderModecv2.BORDER_REPLICATE)returnrotatedexceptException:returnimg注意pdf2image有系统级依赖。这是一个无聊的设置步骤如果跳过它会悄悄降低OCR质量。DPI 300是扫描文档的安全默认值。app/services/ocr.py# app/services/ocr.pyfromPILimportImageimportpytesseractfromtypingimportNamedTupleclassOCRResult(NamedTuple):text:strconfidence:floatdeflocal_tesseract_ocr(image_path:str)-OCRResult:imgImage.open(image_path)datapytesseract.image_to_data(img,output_typepytesseract.Output.DICT)lines[]confidences[]fori,textinenumerate(data.get(text,[])):ifstr(text).strip():lines.append(text)try:conf_valdata.get(conf,[])[i]conffloat(conf_val)ifconf0:confidences.append(conf)exceptException:continuetext\n.join(lines)avg_conf(sum(confidences)/len(confidences)/100.0)ifconfidenceselse0.0returnOCRResult(texttext,confidenceavg_conf)注意实现提供商适配器例如某机构的Textract或Vision LLM返回相同的OCRResult接口。这使得管道的其余部分与提供商无关。app/services/extractor.py — 适配器 稳健的JSON提取# app/services/extractor.pyimportjson,refromapp.configimportsettingsdefextract_json_from_text(text:str)-str:textre.sub(r(?:json)?,,text)brace_idxtext.find({)ifbrace_idx-1:raiseValueError(No JSON object found in response)stack0start-1fori,chinenumerate(text[brace_idx:],startbrace_idx):ifch{:ifstart-1:starti stack1elifch}:stack-1ifstack0:json_strtext[start:i1]try:parsedjson.loads(json_str)returnjson.dumps(parsed)exceptException:breakmre.search(r\{.*\},text,flagsre.DOTALL)ifm:returnm.group(0)raiseValueError(Could not extract JSON from response)# Provider adapter example (conceptual)classOpenAIAdapter:def__init__(self,client):self.clientclientdefextract_with_schema(self,text:str,schema:dict)-dict:# Prefer function-calling / structured outputs if provider supports it.resp_rawself.client.call_model(text,schemaschema,temperature0.0)json_strextract_json_from_text(resp_raw)returnjson.loads(json_str)注意当提供商支持时始终优先使用结构化输出函数调用。自由文本解析之所以存在只是因为真实的模型仍然会以令人惊讶的方式失败。app/services/validator.py — 规范化 验证Decimal, dateutil# app/services/validator.pyfromapp.models.invoiceimportInvoicefrompydanticimportValidationErrorfromdateutilimportparserasdate_parserfromdecimalimportDecimalimportredefnormalize_numbers_and_dates(raw:dict)-dict:amountraw.get(total_amount)ifisinstance(amount,str):amountamount.replace(,,).strip()amountre.sub(r[^\d.\-],,amount)raw[total_amount]Decimal(amount)ifamountelseNoneelifisinstance(amount,(int,float)):raw[total_amount]Decimal(str(amount))date_valraw.get(invoice_date)ifisinstance(date_val,str):try:ddate_parser.parse(date_val,dayfirstFalse)raw[invoice_date]d.date().isoformat()exceptException:passreturnrawdefvalidate_invoice(raw:dict)-Invoice:rawnormalize_numbers_and_dates(raw)try:invoiceInvoice.model_validate(raw)exceptValidationErrorase:raiseRuntimeError(fValidation failed:{e})returninvoice注意在生产环境中明确处理区域设置差异例如日期格式和小数点分隔符。app/services/decision.py — 确定性策略# app/services/decision.pyfromdecimalimportDecimalfromapp.configimportsettings APPROVAL_THRESHOLDDecimal(str(settings.APPROVAL_THRESHOLD))defdecide(invoice):ifinvoice.total_amountAPPROVAL_THRESHOLD:return{decision:AUTO_APPROVED,reason:Amount under threshold,policy_version:v1}return{decision:NEEDS_REVIEW,reason:Amount exceeds threshold,policy_version:v1}重要每个决策都持久化policy_version。当财务或合规部门问“为什么批准了这个”这个字段是你唯一可以辩护的答案。app/db/repo.py — 简化持久化概念性# app/db/repo.pyfromsqlalchemyimportcreate_engine,textfromsqlalchemy.ormimportsessionmakerfromapp.configimportsettings enginecreate_engine(settings.DATABASE_URL)Sessionsessionmaker(bindengine)defsave_document(document_id,path,idempotency_keyNone,content_hashNone):# Implement insert with unique constraints and transactionspassdefsave_process_result(document_id,result_json,provider_meta):# store JSONB recordpassdefget_document_by_idempotency(key):# return document row if existspass确保存储库包含完整的SQL/ORM实现和Alembic迁移见下文以获得完整、可运行的设置。数据库DDLPostgres — infra/db_schema.sqlCREATETABLEdocuments(document_idTEXTPRIMARYKEY,pathTEXTNOTNULL,statusTEXTNOTNULL,idempotency_keyTEXTUNIQUE,content_hashTEXTUNIQUE,created_at TIMESTAMPTZDEFAULTnow());CREATETABLEprocess_results(id BIGSERIALPRIMARYKEY,document_idTEXTREFERENCESdocuments(document_id),result_json JSONBNOTNULL,provider_meta JSONB,created_at TIMESTAMPTZDEFAULTnow());CREATETABLEreview_queue(id BIGSERIALPRIMARYKEY,document_idTEXT,reasonTEXT,added_at TIMESTAMPTZDEFAULTnow(),resolvedBOOLEANDEFAULTFALSE);提示对result_json和provider_meta使用JSONB来存储所有相关元数据包括llm_provider、model、revision和prompt_hash。可观测性快速指南结构化日志使用structlog发出JSON日志包括trace_id、document_id、step和status。指标Prometheus跟踪计数器如documents_ingested_total和documents_processed_total并使用直方图记录processing_duration_seconds。追踪使用OpenTelemetry检测FastAPI和数据库导出到OTLP收集器。在日志中包含trace_id以实现端到端的可追溯性。Alembic示例迁移框架alembic/versions/001_create_tables.pyfromalembicimportopimportsqlalchemyassa revision001defupgrade():op.create_table(documents,sa.Column(document_id,sa.Text,primary_keyTrue),sa.Column(path,sa.Text,nullableFalse),sa.Column(status,sa.Text,nullableFalse),sa.Column(idempotency_key,sa.Text,nullableTrue),sa.Column(content_hash,sa.Text,nullableTrue),sa.Column(created_at,sa.TIMESTAMP(timezoneTrue),server_defaultsa.text(now())))# create other tables...defdowngrade():op.drop_table(documents)人工干预审核端点 — 框架GET /reviews/queue— 获取待审核文档列表POST /reviews/{document_id}/resolve— 提交带有 { decision, note, override_by } 的操作记录每次审核操作附带user_id和时间戳以便审计。UI显示原始PDF、页面级OCR结果、带有置信度得分的提取JSON并提供操作按钮批准、拒绝和编辑。数据保留和删除合规性实现DELETE /documents/{id}作为软删除将文档标记为已删除并在保留TTL生存时间后安排数据块清除。始终尊重法律保留要求并维护完整的删除审计追踪。必须包含的测试验证器测试覆盖日期格式和货币格式的所有边界情况。提取器测试稳健地解析JSON包括围栏式代码块和额外注释。端到端测试使用提供商模拟器来模拟上传 → 处理 → 数据库断言。安全测试验证RBAC端点和权限。快速本地运行启动应用docker-compose up --build上传示例发票curl -F filetests/fixtures/invoice_sample.pdf http://localhost:8000/ingest通过在process_results表中检查JSON和provider_meta来验证结果。爬取 /metrics 端点Prometheus查看计数器是否递增。清单上传使用流式传输不要将整个文件加载到内存中。一致使用Invoice.model_validate / model_dump。对total_amount使用Decimal并稳健地处理日期解析。确保提供商适配器存在并在测试中正确模拟。在content_hash和idempotency_key上应用唯一约束。为审计持久化原始OCR和大语言模型输出。包含Prometheus /metrics和示例仪表板。Alembic迁移应被包含并经过测试。记录后台处理/队列模式。潜在陷阱扫描图像质量极差OCR可能会失败。考虑添加Vision LLM或云端OCR以获得更好的结果。模糊的日期格式DD/MM vs MM/DD确保你的策略明确选择一个区域设置。意外的大语言模型输出结构如果可用使用函数调用或架构强制执行或者进行防御性解析并记录原始输出。注意这些不是错误——它们是你在生产的头几周内会遇到的操作现实。总结本文像为队友编写代码一样编写小巧、清晰、可靠的组件附带审计追踪和操作控制。它很紧凑所以你可以快速掌握构建生产就绪AI系统的关键实践这些系统是安全、可审计和可维护的。将此视为你进行真实世界AI工程的实用蓝图。更多精彩内容 请关注我的个人公众号 公众号办公AI智能小助手或者 我的个人博客 https://blog.qife122.com/对网络安全、黑客技术感兴趣的朋友可以关注我的安全公众号网络安全技术点滴分享