OpenShift AI + DeepSeek+IBM CP4BA：实现AI业务自动化中台

一、OpenShift AI + DeepSeek

1、什么是 OpenShift AI？

OpenShift AI（前身为 Red Hat Open Data Hub）是 Red Hat OpenShift 平台上的一套 AI/ML 工具集，基于 Kubernetes，为企业提供可扩展的机器学习平台，可用于模型的训练、部署、监控和生命周期管理。

核心能力包括：

• GPU 加速支持

• JupyterHub / Notebook 集成

• Pipelines 支持自动化训练流程

• 多用户和权限管理

• 与 Ceph、S3 等存储系统兼容

• 与 GitOps、MLOps 工具链集成 

2、什么是 DeepSeek 大模型？

DeepSeek 是一个开源大语言模型（LLM）系列，类似于 LLaMA、ChatGLM、Baichuan 等。其特点是：

• 模型规模从 7B 到 67B 不等

• 预训练 + 指令微调（Instruction Tuning）

• 开放源码，适合私有化部署

• 支持中英文、代码理解等任务

• 可在本地服务器/GPU 集群上运行

DeepSeek 适合搭建企业级 私有 GPT、知识问答、代码助手、客服 AI 等场景。

3、OpenShift AI + DeepSeek 私有部署组合架构

部署架构大致如下：

|----------------------------|
|     Red Hat OpenShift      |
|----------------------------|
|  OpenShift AI (KubeFlow,   |
|  JupyterHub, ML Pipelines) |
|----------------------------|
|  GPU Nodes (NVIDIA A100)   |
|  Model Serving (KServe)    |
|----------------------------|
|     DeepSeek LLM 模型      |
|    (通过 vLLM / TGI 部署)   |
|----------------------------|
|     企业私有数据集成        |
|     (向量数据库+RAG)        |
|----------------------------|

 关键组件说明：

• 模型加载：通过 vLLM 或 TGI（Text Generation Inference）高性能加载 DeepSeek 模型

• 服务化部署：用 KServe 或 Triton 做模型服务

• 向量数据库：如 Milvus, Weaviate, Qdrant 实现知识增强（RAG）

• 私有数据对接：可以集成企业文档、数据库、搜索系统 

4、优势与特点

优势	说明
💼 私有化部署	数据不出本地，符合企业安全合规要求
⚙️ 企业级支持	OpenShift 提供运维、权限、扩展能力
🚀 高性能推理	DeepSeek + vLLM 提供高并发/低延迟响应
🔗 可扩展	可接入多种数据源、第三方系统
🛠 可控性强	模型版本、资源使用、权限等高度可控

 5、典型应用场景

1. 企业内部知识问答系统（RAG）

2. 专属代码助手 / IDE Copilot

3. 客户服务自动化（ChatBot）

4. 智能文档生成 / 摘要提取

5. 制造/金融/能源行业的 NLP 任务

二、实现对IBM CP4BA的智能化业务中台可行性分析

1、目标场景

通过部署 OpenShift AI + DeepSeek，让用户以 自然语言（聊天）交互的方式 实现对 IBM CP4BA 的以下功能：

• 🔍 基于文档的问答和分析（ECM 如 FileNet）

• 🔄 对业务流程的理解与执行（BPM，如 BAW/BA Studio 流程）

• ✅ 完成流程任务（例如审批、数据录入）

• 📊 对 ECM 文档或业务流程的数据进行分析、汇总、总结

本质上，是将 CP4BA 的“流程 + 内容”能力，通过 AI 对话接口做智能化封装。

2、技术实现路径

下面是一个较为成熟的技术组合（可私有部署）：

1. 打通数据接口（对接 CP4BA）

需要做两类对接：

• ECM 内容：通过 FileNet Content API 获取文档内容、元数据（可转向文本做 RAG 问答）

• BPM 任务/流程数据：调用 CP4BA REST API（包括任务列表、表单数据、流程状态等）

2. 搭建大模型问答能力（用 DeepSeek + RAG）

• 将 ECM 文档做向量化（用 LangChain + FAISS 或 Milvus）

• 通过大模型（DeepSeek）实现 RAG 问答：将用户自然语言转成结合知识的 Prompt

• 可加上 prompt 模板，支持指令式操作如“帮我审批一下XXX流程”

3. 搭建聊天交互系统

前端可使用：

• 基于 ChatUI / Vue / React 的聊天界面

• 嵌入在 CP4BA Portal、BPM Task UI 或独立 Web App

后端：

• Chat Controller：解析用户意图

• 动作执行器：根据模型返回意图执行 API 调用，如：

  • 查询某流程状态

  • 审批某任务

  • 下载某文档内容

  • 汇总某个客户对应的所有业务记录

4. 模型服务部署（OpenShift AI + DeepSeek）

• 使用 OpenShift AI 提供的 GPU 支持，部署 DeepSeek 模型服务（推荐用 vLLM + Triton）

• 部署 LLM 服务组件 + 向量搜索引擎 + 控制服务 

3、参考架构图（简略）

[用户聊天界面]│▼
[聊天解析层 - 意图识别 / RAG]│┌───────────────┐│DeepSeek 模型服务│  ← OpenShift AI 上运行└───────────────┘│┌──────────────┐      ┌──────────────┐│ECM 适配器     │←REST→│ FileNet 系统   │└──────────────┘      └──────────────┘│┌──────────────┐      ┌──────────────┐│BPM 适配器     │←REST→│ CP4BA 流程引擎 │└──────────────┘      └──────────────┘

4、示例对话能力

用户问题	AI 能力
“请帮我查一下张三的贷款流程审批状态”	解析“张三+贷款流程”，调用 BPM 查询 API
“我能看看他提交的资料吗？”	通过关联流程和 FileNet ECM，提取文档内容
“帮我把这个流程提交给上级”	模型返回意图 → 执行流程任务 API 提交
“总结一下我们这个月处理的所有投诉流程”	汇总 BPM 流程 + 结合模型生成报告

5、关键组件/技术推荐

模块	技术栈推荐
模型服务	DeepSeek-7B/67B + vLLM/TGI
模型部署平台	OpenShift AI（支持 GPU & Operator 管理）
向量数据库	Milvus / Weaviate / Qdrant
文档解析	Apache Tika / pdfminer / OCR
对话管理	LangChain / Semantic Router
API 对接	FileNet/BAW REST API
安全认证	OAuth2 / CP4BA SSO 接入

6、可落地性评估

需求是完全可落地的，并且 IBM 自家的 CP4BA 本身就是云原生，和 OpenShift 配合度非常高。DeepSeek 模型也支持私有部署，非常适合放在本地 GPU 集群中与 OpenShift AI 集成使用。

三、 POC 实施方案：智能 AI 聊天访问 CP4BA（ECM + BPM）

做 POC（概念验证）是推动这类项目落地的最佳方式。

1、POC目标

通过部署 OpenShift AI + DeepSeek 模型，构建一个原型系统，实现以下 AI 能力：

• ✅ 实现对 ECM 文档的自然语言问答与总结（RAG 模式）

• ✅ 实现对 BPM 流程的查询、控制、任务处理（自然语言转接口调用）

• ✅ 构建端到端聊天系统，打通用户 → 模型 → CP4BA 的操作链路

• ✅ 在 OpenShift AI 上私有化部署大语言模型 DeepSeek

2、POC 架构总览

┌──────────────────────────────────────┐
│            Chat Web 前端              │
│  (React/Vue + ChatUI)               │
└────────────▲─────────────────────────┘│▼
┌──────────────────────────────────────┐
│     对话控制器（LangChain, FastAPI）   │
│ - 用户意图识别                        │
│ - Prompt 构建 / 模型调用              │
│ - 动作指令转 REST API                │
└────▲─────────────▲───────────────┘│             ││             │▼             ▼
┌────────────┐   ┌────────────────────┐
│ DeepSeek LLM │   │ 向量数据库（RAG）       │
│（vLLM部署） │   │ FAISS / Milvus         │
└────▲───────┘   └────▲─────────────────┘│                │▼                ▼
┌────────────┐    ┌────────────┐
│ ECM Adapter │    │ BPM Adapter │
│ (FileNet API)│    │ (BAW REST API)│
└────▲───────┘    └────▲───────┘│                │▼                ▼
┌────────────┐    ┌────────────┐
│ FileNet 文档库 │    │ BAW 流程引擎 │
└────────────┘    └────────────┘

3、核心组件及技术选型建议

模块	技术选型
模型服务	DeepSeek-7B Chat + vLLM
向量数据库	FAISS 或 Milvus
模型部署平台	Red Hat OpenShift AI（GPU Operator 支持）
对话框架	LangChain（Python）
前端聊天界面	ChatUI + Vue/React
数据源适配	FileNet Content API / BAW REST API
权限认证	内网部署，集成企业 OAuth2 / CP4BA 登录体系

4、POC范围定义

1. 文档问答功能（RAG）

• 文档嵌入：选取典型 FileNet 文档 → 提取文本 → 向量化入库

• 用户提问：如“张三贷款文档中，担保人是谁？” → 模型结合上下文回答

 文档向量化脚本（PDF / Word / TXT）

from langchain.document_loaders import UnstructuredFileLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
import os# 文档路径，可循环多个文档
doc_dir = "docs/"
all_chunks = []# 遍历并加载文件
for file_name in os.listdir(doc_dir):loader = UnstructuredFileLoader(os.path.join(doc_dir, file_name))docs = loader.load()splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)chunks = splitter.split_documents(docs)all_chunks.extend(chunks)# 构建向量数据库
embedding = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
vector_db = FAISS.from_documents(all_chunks, embedding)# 保存索引
vector_db.save_local("vector_index/")

2. 流程查询与控制

• 查询任务状态：“张三的贷款流程到哪一步了？”

• 任务处理：“帮我审批这个流程” → 模型识别意图 → 执行 BPM API 提交任务

 FileNet + BAW 的 API 适配器代码

import requestsclass BAWAdapter:def __init__(self, base_url, username, password):self.base_url = base_url.rstrip('/')self.session = requests.Session()self.session.auth = (username, password)def get_user_tasks(self, user_id):url = f"{self.base_url}/rest/bpm/wle/v1/tasks?assignedTo={user_id}"resp = self.session.get(url)return resp.json()def complete_task(self, task_id, params=None):url = f"{self.base_url}/rest/bpm/wle/v1/task/{task_id}?action=finish"resp = self.session.put(url, data=params or {})return resp.json()

 示例调用：

baw = BAWAdapter("http://baw-server:9080", "admin", "passw0rd")
tasks = baw.get_user_tasks("zhangsan")
print(tasks)

3. 数据分析 / 总结能力

• “总结下本月处理的投诉流程” → 查询 + 汇总 + AI 生成总结文本

实现：

• AI 自动从 FileNet / BAW 获取结构化数据

• 汇总/分析后生成图表（柱状图、折线图、饼图）

• 结果以图 + 文总结形式返回前端聊天界面

LangChain 调用“数据分析工具”

使用 Tool 或 Agent 机制，注册一个 DataQueryTool：

from langchain.agents import Tooldef query_complaint_summary(time_range: str):# 调用 BAW 查询 API，获取流程实例列表，按类别统计return {"投诉类别A": 12,"类别B": 9,"类别C": 5}tools = [Tool(name="ComplaintSummaryTool",func=query_complaint_summary,description="查询指定时间段的客户投诉流程统计")
]

 服务端生成图表（可选用 Matplotlib / Plotly）

import matplotlib.pyplot as pltdef generate_bar_chart(data: dict, title: str = ""):fig, ax = plt.subplots()ax.bar(data.keys(), data.values(), color='skyblue')ax.set_title(title)ax.set_ylabel("数量")ax.set_xlabel("类别")fig.tight_layout()# 保存图片filename = "chart.png"fig.savefig(filename)return filename

 聊天接口中返回图表

FastAPI 后端返回格式：

{"type": "chart","image_url": "/static/chart.png","caption": "2025年4月客户投诉流程统计图"
}

 前端渲染图表消息（React 示例）

{message.type === "chart" && (<div className="chat-message bot"><img src={message.image_url} alt="分析图表" className="rounded-lg w-80" /><p className="text-sm text-gray-500 mt-1">{message.caption}</p></div>
)}

支持的图表类型建议

类型	场景
柱状图	各类流程数量 / 文档量
折线图	时间趋势分析
饼图	占比统计（例如投诉类别）
表格	多字段对比

4. 聊天交互界面

• 搭建对话 Web UI，支持输入、对话记录、上下文保持、错误提示等

示例前端代码（React + Tailwind）

import React, { useState } from 'react';export default function ChatInterface() {const [messages, setMessages] = useState([]);const [input, setInput] = useState('');const sendMessage = async () => {const userMessage = { role: 'user', content: input };setMessages([...messages, userMessage]);setInput('');const res = await fetch('/api/chat', {method: 'POST',headers: { 'Content-Type': 'application/json' },body: JSON.stringify({ message: input })});const data = await res.json();const botMessage = { role: 'assistant', content: data.reply };setMessages((prev) => [...prev, botMessage]);};return (<div className="w-full h-screen p-4 bg-gray-50 flex flex-col"><div className="flex-1 overflow-y-auto mb-4">{messages.map((m, i) => (<div key={i} className={`my-2 p-2 rounded-lg ${m.role === 'user' ? 'bg-blue-100 text-right' : 'bg-gray-200 text-left'}`}>{m.content}</div>))}</div><div className="flex gap-2"><inputtype="text"className="flex-1 p-2 rounded border"value={input}onChange={(e) => setInput(e.target.value)}onKeyDown={(e) => e.key === 'Enter' && sendMessage()}/><button className="px-4 py-2 bg-blue-500 text-white rounded" onClick={sendMessage}>发送</button></div></div>);
}

示例后端代码（FastAPI + LangChain）

from fastapi import FastAPI, Request
from pydantic import BaseModel
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA
from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddingsapp = FastAPI()# 加载向量数据库
db = FAISS.load_local("vector_index/", HuggingFaceEmbeddings())
qa_chain = RetrievalQA.from_chain_type(llm=OpenAI(base_url="http://deepseek-service:8000/v1"), retriever=db.as_retriever())class ChatRequest(BaseModel):message: str@app.post("/api/chat")
def chat(req: ChatRequest):response = qa_chain.run(req.message)return {"reply": response}

5. OAuth2 + CP4BA 登录整合方案

企业用户通过浏览器访问聊天系统前端时，将自动跳转至企业统一身份认证平台（如 Keycloak / SSO 网关）进行登录，登录成功后携带 JWT Token 请求后端服务。

后端统一身份认证机制可支持以下两种方式：

✅ 模式 A：OAuth2 JWT 校验（推荐）

适用于前端获取 JWT Token 后传入后端，后端进行解码与验证。

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTErroroauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")SECRET_KEY = "your-oauth2-jwt-secret"
ALGORITHM = "HS256"  # 或 RS256 视实际情况而定async def get_current_user(token: str = Depends(oauth2_scheme)):try:payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])return payload  # 包含用户名、角色等except JWTError:raise HTTPException(status_code=401, detail="Invalid token")

✅ 模式 B：CP4BA 内置 LTPA Token（适用于旧系统）

如系统已启用 WebSphere LTPA 登录机制，可从 Cookie 中解析用户身份。

5、所需资源（基础配置）

1. 软件环境

• OpenShift AI 已部署集群

• DeepSeek 模型本地文件（7B）

• CP4BA 环境（含 FileNet 和 BAW）

• 向量数据库服务（容器化部署）

2. 硬件资源（推荐最小配置）

• GPU：A10×2 或 A100×1

• CPU：16 Core+

• RAM：64GB+

• 存储：500GB+

 6、DeepSeek 模型部署示例（vLLM + OpenShift AI）

# openshift-deepseek-llm.yaml（部署 vLLM 作为服务）
apiVersion: serving.knative.dev/v1
kind: Service
metadata:name: deepseek-7bnamespace: ai-models
spec:template:spec:containers:- name: vllm-deepseekimage: vllm/vllm-openai:latestcommand: ["python3", "-m", "vllm.entrypoints.openai.api_server"]args:- "--model"- "/models/deepseek-chat-7b"- "--tokenizer"- "deepseek-ai/deepseek-7b"- "--dtype"- "bfloat16"- "--port"- "8000"resources:limits:nvidia.com/gpu: 1memory: "48Gi"cpu: "8"volumeMounts:- name: model-volumemountPath: /modelsvolumes:- name: model-volumepersistentVolumeClaim:claimName: deepseek-model-pvc

配合 vLLM，将模型以 OpenAI-compatible API 暴露出来，供 LangChain 调用。

7、文档向量化 + RAG 入库脚本（Python 示例）

import requestsdef get_filenet_doc(doc_id, access_token):url = f"https://filenet.company.com/api/documents/{doc_id}/content"headers = {"Authorization": f"Bearer {access_token}"}response = requests.get(url, headers=headers)if response.ok:with open(f"{doc_id}.pdf", "wb") as f:f.write(response.content)else:print("Failed:", response.status_code)# 示例
get_filenet_doc("ecm-12345", "your-access-token")

 📌 可用于向量化 ECM 文档（建议抽取 FileNet 文档内容后保存为 PDF/HTML 处理）。

8、可选：调用 BAW 查询流程状态

def get_baw_process(process_id, token):url = f"https://baw.company.com/rest/bpm/wle/v1/process/{process_id}"headers = {"Authorization": f"Bearer {token}"}res = requests.get(url, headers=headers)return res.json()

 这些代码示例可以直接嵌入到 LangChain 或 Flask/FastAPI 对话服务中。如果使用的是企业 VPN、SSO 环境，也可添加 CP4BA 用户凭据校验逻辑。

四、聊天式数据分析与图表展示模块

1、模块结构（模块名：data_analysis_toolkit）

data_analysis_toolkit/
├── __init__.py
├── tools.py              # LangChain Tool 注册
├── charting.py           # 图表绘制模块（柱状图/饼图/折线图）
├── data_fetcher.py       # FileNet / BAW 数据抓取逻辑
├── prompts.py            # 分析提示词模板（可选）
├── ecm_fetcher.py        # 新增：封装 FileNet 文档查询逻辑
├── tools.py              # 扩展：新增 ECM 文档图表工具

2、ecm_fetcher.py 示例（模拟 FileNet 查询）

可以用真实 API 替换这个模拟逻辑：

from datetime import datetime, timedeltadef fetch_ecm_docs_by_day(days=7) -> dict:# 模拟：最近 N 天每天的文档数量today = datetime.today()return {(today - timedelta(days=i)).strftime("%m-%d"): (10 + i*3) % 20for i in reversed(range(days))}def fetch_ecm_docs_by_type() -> dict:return {"合同": 42,"发票": 33,"报告": 27,"其他": 14}def fetch_user_docs_trend(user: str, days=7) -> dict:return {f"{i+1}天前": (5 + i * 2) % 10for i in reversed(range(days))}

3、tools.py – LangChain Tool 封装

from langchain.agents import Tool
from .data_fetcher import fetch_complaint_data
from .charting import generate_bar_chartdef complaint_summary_tool(time_range: str):data = fetch_complaint_data(time_range)img_path = generate_bar_chart(data, title="投诉流程统计")return {"caption": f"{time_range} 客户投诉流程数量统计","image_url": f"/static/{img_path}"}tools = [Tool(name="ComplaintSummary",func=complaint_summary_tool,description="获取指定时间段内的投诉流程数量并绘图")
]

 注册 LangChain 工具

from .ecm_fetcher import fetch_ecm_docs_by_day, fetch_ecm_docs_by_type, fetch_user_docs_trend
from .charting import generate_bar_chart, generate_pie_chartdef ecm_trend_tool(_):data = fetch_ecm_docs_by_day()path = generate_bar_chart(data, "近7天新增文档数量趋势")return {"caption": "近7天新增文档趋势图","image_url": f"/static/{path}"}def ecm_type_dist_tool(_):data = fetch_ecm_docs_by_type()path = generate_pie_chart(data, "文档类型分布")return {"caption": "当前文档类型分布图","image_url": f"/static/{path}"}def user_upload_trend(user: str):data = fetch_user_docs_trend(user)path = generate_bar_chart(data, f"{user} 上传文档趋势")return {"caption": f"{user} 最近上传文档趋势图","image_url": f"/static/{path}"}tools.extend([Tool(name="ECMTrend", func=ecm_trend_tool, description="查看文档上传趋势图"),Tool(name="ECMTypeDist", func=ecm_type_dist_tool, description="查看文档类型分布图"),Tool(name="UserECMTrend", func=user_upload_trend, description="查看用户上传趋势")
])

4、charting.py – 图表绘制模块 

import matplotlib.pyplot as plt
import uuid
import osdef generate_bar_chart(data: dict, title: str):fig, ax = plt.subplots()ax.bar(data.keys(), data.values(), color='skyblue')ax.set_title(title)ax.set_ylabel("数量")ax.set_xlabel("分类")filename = f"chart_{uuid.uuid4().hex[:6]}.png"path = os.path.join("static", filename)fig.tight_layout()fig.savefig(path)plt.close(fig)return filename

5、 图表函数补充（generate_pie_chart）

def generate_pie_chart(data: dict, title: str):fig, ax = plt.subplots()ax.pie(data.values(), labels=data.keys(), autopct='%1.1f%%', startangle=90)ax.axis('equal')plt.title(title)filename = f"chart_{uuid.uuid4().hex[:6]}.png"path = os.path.join("static", filename)plt.savefig(path)plt.close(fig)return filename

6、data_fetcher.py – 数据源接入（BAW 示例） 

import requestsdef fetch_complaint_data(time_range: str) -> dict:# 示例逻辑：实际应根据 time_range 查询 BAW 流程# 假设返回模拟数据return {"产品质量": 14,"服务态度": 6,"物流问题": 4,"其他": 2}# 可扩展：支持 FileNet 文档统计 / 其他类型流程分析

7、对接聊天接口（FastAPI） 

from data_analysis_toolkit.tools import tools
from langchain.agents import initialize_agent
from langchain.chat_models import ChatOpenAIllm = ChatOpenAI(base_url="http://localhost:8000/v1", model="deepseek")
agent = initialize_agent(tools, llm, agent_type="zero-shot-react-description")@app.post("/api/chat")
def chat_with_ai(user_input: str):result = agent.run(user_input)if isinstance(result, dict) and "image_url" in result:return {"type": "chart","image_url": result["image_url"],"caption": result["caption"]}else:return {"type": "text", "content": result}

8、效果展示（聊天界面）

用户输入：

“展示一下3月份各类客户投诉流程数量的分布图”

响应：

• 图表（柱状图或饼图）

• 标题：“3月份客户投诉流程统计图”

• 简要文字总结：“投诉主要集中在产品质量类，共14件，占比60%。”

🧑‍💼 用户：展示一下这周新增了多少文档？
🤖 AI：这是最近7天的文档上传数量趋势👇
[📊 图表展示]
共上传文档 86 份，其中周三最多，达到了 19 份。

五、FileNet REST API 查询文档列表（Python 示例）

IBM FileNet Content Platform Engine (CPE) 的文档数据分析接口封装代码支持通过 REST API 获取文档信息，并用于图表分析与聊天问答中。

1、接口方案说明

你可以选择使用：

1. CMIS 接口（标准，适配多系统，但能力有限）

2. ✅ FileNet REST API（推荐）：灵活，可直接查询文档类、自定义属性、时间过滤等。

2、示例：获取某类文档过去 7 天每天上传的数量 

import requests
from datetime import datetime, timedelta
from collections import defaultdictFILENET_API_BASE = "https://your-cp4ba-host.com/ibm/content-services/api"
AUTH_HEADER = {"Authorization": "Bearer <your_token>","Accept": "application/json"
}def get_docs_by_day(class_name="Document", days=7):counts = defaultdict(int)for i in range(days):start = (datetime.now() - timedelta(days=i+1)).strftime("%Y-%m-%dT00:00:00Z")end = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%dT00:00:00Z")url = f"{FILENET_API_BASE}/v1/objects?objectType={class_name}&query=SELECT * FROM {class_name} WHERE DateCreated BETWEEN TIMESTAMP '{start}' AND TIMESTAMP '{end}'"res = requests.get(url, headers=AUTH_HEADER)if res.ok:date_key = (datetime.now() - timedelta(days=i)).strftime("%m-%d")counts[date_key] = len(res.json().get("documents", []))else:print("❌ Error:", res.status_code, res.text)return dict(sorted(counts.items()))# 示例：
# {'04-06': 12, '04-07': 9, '04-08': 5, ...}

3、示例：按文档类型统计数量 

def get_doc_count_by_type():type_map = {"合同类": "ContractDoc","发票类": "InvoiceDoc","报告类": "ReportDoc"}result = {}for label, class_name in type_map.items():url = f"{FILENET_API_BASE}/v1/objects?objectType={class_name}&query=SELECT Id FROM {class_name}"res = requests.get(url, headers=AUTH_HEADER)result[label] = len(res.json().get("documents", [])) if res.ok else 0return result

4、示例：指定用户上传文档趋势（按 Creator）

def get_user_docs_trend(username, days=7):counts = defaultdict(int)for i in range(days):start = (datetime.now() - timedelta(days=i+1)).strftime("%Y-%m-%dT00:00:00Z")end = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%dT00:00:00Z")query = (f"SELECT Id FROM Document WHERE Creator = '{username}' AND "f"DateCreated BETWEEN TIMESTAMP '{start}' AND TIMESTAMP '{end}'")url = f"{FILENET_API_BASE}/v1/objects?objectType=Document&query={query}"res = requests.get(url, headers=AUTH_HEADER)key = (datetime.now() - timedelta(days=i)).strftime("%m-%d")counts[key] = len(res.json().get("documents", [])) if res.ok else 0return dict(sorted(counts.items()))

5、认证方式说明

• 推荐：使用 CP4BA 的 OAuth Token（通过 SSO 或 API Gateway 获取）

• 也支持 Basic Auth（开发环境）

AUTH_HEADER = {"Authorization": f"Basic base64(username:password)"
}

6、集成图表模块

可以将以上函数分别接入：

图表	函数
柱状图 - 文档日趋势	get_docs_by_day()
饼图 - 文档类型占比	get_doc_count_by_type()
折线图 - 用户上传趋势	get_user_docs_trend(username)

返回的数据结构可以直接喂给之前构建的 generate_bar_chart() 或 generate_pie_chart()。

7、示例图表效果（文本描述）

📊 图表：近7天合同类文档上传数量日期   数量  
04-05   3  
04-06   5  
04-07   8  
04-08   6  
04-09   9  
04-10   10  
04-11   4  文档数量整体呈上升趋势，上传高峰集中在 4月9~10日。