From c9bebe56159b27fd3f5e1e78f559041742605ddd Mon Sep 17 00:00:00 2001 From: chenjw28 <792430652@qq.com> Date: Mon, 2 Mar 2026 17:41:04 +0800 Subject: [PATCH] init --- Jenkinsfile | 299 ++++++++++++++++++++++++++------ README.md | 9 +- api/endpoints.py | 31 ++-- schemas/agent_input.py | 17 +- schemas/chat_message_request.py | 22 +++ schemas/stream_input.py | 7 + scripts/test_endpoints.py | 23 ++- 7 files changed, 322 insertions(+), 86 deletions(-) create mode 100644 schemas/chat_message_request.py create mode 100644 schemas/stream_input.py diff --git a/Jenkinsfile b/Jenkinsfile index 8bf316d..149a384 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -1,64 +1,255 @@ -stage('Deploy to k3s') { - steps { - dir('more_dots') { - script { - def cluster = params.TARGET_CLUSTER - def fullImageWithTag = "${FULL_IMAGE_NAME}:${env.IMAGE_TAG}" +pipeline { + agent any - if (cluster == 'cluster1' || cluster == 'both') { - withCredentials([ - file(credentialsId: 'k3s-cluster1-config', variable: 'KUBECONFIG_CLUSTER1'), - usernamePassword(credentialsId: REGISTRY_CREDENTIALS_ID, usernameVariable: 'REGISTRY_USER', passwordVariable: 'REGISTRY_PASS') - ]) { + environment { + // GitLab配置 + GITLAB_URL = 'https://gitlab.xpaas.lenovo.com' + GITLAB_REPO = 'artificial-intelligence-platform/lenovo-ipc' + GITLAB_CREDENTIALS_ID = 'more_dots' + + // 新镜像仓库配置 (10.128.62.130:8843) + REGISTRY_URL = '10.128.62.130:8843' + REGISTRY_CREDENTIALS_ID = 'docker-registry-130' + + // 镜像配置 + IMAGE_NAME = 'more_dots' + FULL_IMAGE_NAME = "${REGISTRY_URL}/${IMAGE_NAME}" + + // 是否跳过构建(运行时赋值) + SKIP_BUILD = 'false' + } + + parameters { + choice(name: 'DEPLOY_ENV', choices: ['dev', 'staging', 'prod'], description: '选择部署环境') + choice(name: 'TARGET_CLUSTER', choices: ['cluster1', 'cluster2', 'both'], description: '选择目标集群') + } + + stages { + stage('Checkout') { + steps { + git( + url: "${GITLAB_URL}/${GITLAB_REPO}.git", + credentialsId: GITLAB_CREDENTIALS_ID, + branch: 'master' + ) + script { + env.GIT_COMMIT = sh(script: 'git rev-parse HEAD', returnStdout: true).trim() + env.SOURCE_TREE = sh(script: 'git rev-parse HEAD:more_dots', returnStdout: true).trim() + env.IMAGE_TAG = env.SOURCE_TREE.take(12) + + echo "GIT_COMMIT: ${env.GIT_COMMIT}" + echo "SOURCE_TREE: ${env.SOURCE_TREE}" + echo "IMAGE_TAG: ${env.IMAGE_TAG}" + echo "FULL_IMAGE_NAME: ${FULL_IMAGE_NAME}:${env.IMAGE_TAG}" + } + } + } + + stage('Check Code Changes') { + steps { + script { + def prev = env.GIT_PREVIOUS_SUCCESSFUL_COMMIT ?: '' + + if (!prev?.trim()) { + echo "未找到上次成功构建提交,默认执行构建" + env.SKIP_BUILD = 'false' + } else { + int rc = sh( + script: "git diff --quiet ${prev} HEAD -- more_dots", + returnStatus: true + ) + env.SKIP_BUILD = (rc == 0) ? 'true' : 'false' + } + + echo "代码变更检查结果: SKIP_BUILD=${env.SKIP_BUILD}" + } + } + } + + stage('Build Docker Image') { + when { + expression { env.SKIP_BUILD != 'true' } + } + steps { + dir('more_dots') { + sh "docker build --progress=plain -t ${IMAGE_NAME}:${env.IMAGE_TAG} ." + } + } + } + + stage('Push to Registry') { + when { + expression { env.SKIP_BUILD != 'true' } + } + steps { + script { + withCredentials([usernamePassword( + credentialsId: REGISTRY_CREDENTIALS_ID, + usernameVariable: 'REGISTRY_USER', + passwordVariable: 'REGISTRY_PASS' + )]) { sh """ set -eux - - # 查找kubectl路径 - KUBECTL_PATH=\$(command -v kubectl 2>/dev/null || true) - if [ -z "\$KUBECTL_PATH" ]; then - for p in /usr/local/bin/kubectl /usr/bin/kubectl /bin/kubectl; do - if [ -x "\$p" ]; then - KUBECTL_PATH="\$p" - break - fi - done - fi - - echo "使用kubectl路径: \$KUBECTL_PATH" - - # 定义kubectl函数 - k() { - sudo \$KUBECTL_PATH --kubeconfig=${KUBECONFIG_CLUSTER1} "\$@" - } - - # 检查命名空间 - k get namespace ${params.DEPLOY_ENV} || k create namespace ${params.DEPLOY_ENV} - - # 创建imagePullSecret - k create secret docker-registry regcred-130 \\ - --docker-server=${REGISTRY_URL} \\ - --docker-username=${REGISTRY_USER} \\ - --docker-password=${REGISTRY_PASS} \\ - --namespace=${params.DEPLOY_ENV} \\ - --dry-run=client -o yaml | k apply -f - - - # 替换镜像并部署 - sed "s|image:.*more_dots.*|image: ${fullImageWithTag}|g" k8s/deployment.yaml > /tmp/deployment-${params.DEPLOY_ENV}.yaml - k apply -f /tmp/deployment-${params.DEPLOY_ENV}.yaml -n ${params.DEPLOY_ENV} - - # 确保使用imagePullSecret(注意 deployment 名称是 more-dots) - k patch deployment more-dots -n ${params.DEPLOY_ENV} \\ - -p '{"spec":{"template":{"spec":{"imagePullSecrets":[{"name":"regcred-130"}]}}}}' || true - - # 重启并等待 - k rollout restart deployment/more-dots -n ${params.DEPLOY_ENV} 2>/dev/null || true - k rollout status deployment/more-dots -n ${params.DEPLOY_ENV} --timeout=300s + echo "登录到镜像仓库 ${REGISTRY_URL}..." + echo "${REGISTRY_PASS}" | docker login ${REGISTRY_URL} -u ${REGISTRY_USER} --password-stdin """ } - } - // cluster2 部分同样修改... + sh """ + set -eux + echo "打标签: ${IMAGE_NAME}:${IMAGE_TAG} -> ${FULL_IMAGE_NAME}:${IMAGE_TAG}" + docker tag ${IMAGE_NAME}:${IMAGE_TAG} ${FULL_IMAGE_NAME}:${IMAGE_TAG} + + echo "推送镜像到仓库..." + docker push ${FULL_IMAGE_NAME}:${IMAGE_TAG} + + echo "镜像推送完成: ${FULL_IMAGE_NAME}:${IMAGE_TAG}" + """ + } + } + } + + stage('Deploy to k3s') { + steps { + dir('more_dots') { + script { + def cluster = params.TARGET_CLUSTER + def fullImageWithTag = "${FULL_IMAGE_NAME}:${env.IMAGE_TAG}" + + if (cluster == 'cluster1' || cluster == 'both') { + withCredentials([ + file(credentialsId: 'k3s-cluster1-config', variable: 'KUBECONFIG_CLUSTER1'), + usernamePassword(credentialsId: REGISTRY_CREDENTIALS_ID, usernameVariable: 'REGISTRY_USER', passwordVariable: 'REGISTRY_PASS') + ]) { + sh """ + set -eux + + # 查找kubectl路径 + KUBECTL_PATH=\$(command -v kubectl 2>/dev/null || true) + if [ -z "\$KUBECTL_PATH" ]; then + for p in /usr/local/bin/kubectl /usr/bin/kubectl /bin/kubectl; do + if [ -x "\$p" ]; then + KUBECTL_PATH="\$p" + break + fi + done + fi + + if [ -z "\$KUBECTL_PATH" ]; then + echo "kubectl 未找到,请检查 Jenkins 节点环境" + exit 1 + fi + + echo "使用kubectl路径: \$KUBECTL_PATH" + + # 定义kubectl函数 + k() { + sudo \$KUBECTL_PATH --kubeconfig=${KUBECONFIG_CLUSTER1} "\$@" + } + + # 检查命名空间 + k get namespace ${params.DEPLOY_ENV} || k create namespace ${params.DEPLOY_ENV} + + # 创建imagePullSecret + k create secret docker-registry regcred-130 \\ + --docker-server=${REGISTRY_URL} \\ + --docker-username=${REGISTRY_USER} \\ + --docker-password=${REGISTRY_PASS} \\ + --namespace=${params.DEPLOY_ENV} \\ + --dry-run=client -o yaml | k apply -f - + + # 替换镜像并部署 + sed "s|image:.*python-app.*|image: ${fullImageWithTag}|g" k8s/deployment.yaml > /tmp/deployment-${params.DEPLOY_ENV}.yaml + k apply -f /tmp/deployment-${params.DEPLOY_ENV}.yaml -n ${params.DEPLOY_ENV} + + # 确保使用imagePullSecret + k patch deployment python-app -n ${params.DEPLOY_ENV} \\ + -p '{"spec":{"template":{"spec":{"imagePullSecrets":[{"name":"regcred-130"}]}}}}' || true + + # 重启并等待 + k rollout restart deployment/python-app -n ${params.DEPLOY_ENV} 2>/dev/null || true + k rollout status deployment/python-app -n ${params.DEPLOY_ENV} --timeout=300s + """ + } + } + + if (cluster == 'cluster2' || cluster == 'both') { + withCredentials([ + file(credentialsId: 'k3s-cluster2-config', variable: 'KUBECONFIG_CLUSTER2'), + usernamePassword(credentialsId: REGISTRY_CREDENTIALS_ID, usernameVariable: 'REGISTRY_USER', passwordVariable: 'REGISTRY_PASS') + ]) { + sh """ + set -eux + + # 查找kubectl路径 + KUBECTL_PATH=\$(command -v kubectl 2>/dev/null || true) + if [ -z "\$KUBECTL_PATH" ]; then + for p in /usr/local/bin/kubectl /usr/bin/kubectl /bin/kubectl; do + if [ -x "\$p" ]; then + KUBECTL_PATH="\$p" + break + fi + done + fi + + if [ -z "\$KUBECTL_PATH" ]; then + echo "kubectl 未找到,请检查 Jenkins 节点环境" + exit 1 + fi + + echo "使用kubectl路径: \$KUBECTL_PATH" + + # 定义kubectl函数 + k() { + sudo \$KUBECTL_PATH --kubeconfig=${KUBECONFIG_CLUSTER2} "\$@" + } + + # 检查命名空间 + k get namespace ${params.DEPLOY_ENV} || k create namespace ${params.DEPLOY_ENV} + + # 创建imagePullSecret + k create secret docker-registry regcred-130 \\ + --docker-server=${REGISTRY_URL} \\ + --docker-username=${REGISTRY_USER} \\ + --docker-password=${REGISTRY_PASS} \\ + --namespace=${params.DEPLOY_ENV} \\ + --dry-run=client -o yaml | k apply -f - + + # 替换镜像并部署 + sed "s|image:.*python-app.*|image: ${fullImageWithTag}|g" k8s/deployment.yaml > /tmp/deployment-${params.DEPLOY_ENV}.yaml + k apply -f /tmp/deployment-${params.DEPLOY_ENV}.yaml -n ${params.DEPLOY_ENV} + + # 确保使用imagePullSecret + k patch deployment python-app -n ${params.DEPLOY_ENV} \\ + -p '{"spec":{"template":{"spec":{"imagePullSecrets":[{"name":"regcred-130"}]}}}}' || true + + # 重启并等待 + k rollout restart deployment/python-app -n ${params.DEPLOY_ENV} 2>/dev/null || true + k rollout status deployment/python-app -n ${params.DEPLOY_ENV} --timeout=300s + """ + } + } + } + } } } } + + post { + always { + script { + sh """ + docker logout ${REGISTRY_URL} || true + docker rmi ${FULL_IMAGE_NAME}:${env.IMAGE_TAG} || true + """ + } + cleanWs() + echo "构建结束。镜像标签: ${env.IMAGE_TAG}" + echo "完整镜像: ${FULL_IMAGE_NAME}:${env.IMAGE_TAG}" + echo "是否跳过构建: ${env.SKIP_BUILD}" + } + failure { + echo "构建失败,请检查日志" + } + } } \ No newline at end of file diff --git a/README.md b/README.md index 06d8f7c..8f4e52f 100644 --- a/README.md +++ b/README.md @@ -125,9 +125,12 @@ manager_alt = WorkflowManager(default_model_section="gpt-3.5-turbo") ```json { - "input": "你好,帮我算 1 + 2", - "session_id": null, - "workflow_type": "conversation" + "query": "你好,帮我算 1 + 2", + "conversation_id": null, + "workflow_type": "conversation", + "response_mode": "blocking", + "user": "demo", + "inputs": {} } ``` diff --git a/api/endpoints.py b/api/endpoints.py index ecf0c38..6e2ef0d 100644 --- a/api/endpoints.py +++ b/api/endpoints.py @@ -12,6 +12,7 @@ from schemas.agent_output import AgentOutput from schemas.tool_input import ToolInput from schemas.tool_output import ToolOutput from schemas.chat_message_response import ChatMessageResponseDTO +from schemas.chat_message_request import ChatMessageRequestDTO from workflows.workflow_manager import WorkflowType from api.dependencies import get_workflow_manager, get_nacos_manager, get_service_config, get_tool_router, get_prompt_manager from services.app_errors import AppError, ErrorCode @@ -68,8 +69,8 @@ def run_workflow(payload: AgentInput, workflow_manager=Depends(get_workflow_mana try: result = workflow_manager.execute_workflow( workflow_type=workflow_type, - user_input=payload.input, - session_id=payload.session_id, + user_input=payload.query, + session_id=payload.conversation_id, ) slog.log("INFO", "run_workflow.success", trace_id, {"session_id": result.get("session_id")}) except Exception as e: @@ -96,8 +97,8 @@ def generate_sql(payload: AgentInput, workflow_manager=Depends(get_workflow_mana result = workflow_manager.execute_workflow( workflow_type=workflow_type, - user_input=payload.input, - session_id=payload.session_id, + user_input=payload.query, + session_id=payload.conversation_id, skip_sr_api=True, ) @@ -118,16 +119,12 @@ def generate_sql(payload: AgentInput, workflow_manager=Depends(get_workflow_mana @router.post("/api/workflows/stream") -def run_workflow_stream(payload: AgentInput, workflow_manager=Depends(get_workflow_manager)): +def run_workflow_stream(payload: ChatMessageRequestDTO, workflow_manager=Depends(get_workflow_manager)): trace_id = uuid.uuid4().hex slog = get_structured_logger() - try: - workflow_type = _resolve_workflow_type(payload.workflow_type) - except Exception as e: - raise _to_http_error(e) - if workflow_type != WorkflowType.CONVERSATION: - raise _to_http_error(AppError(code=ErrorCode.INVALID_WORKFLOW_TYPE, message="仅支持对话工作流的流式输出", status_code=400)) + if payload.response_mode != "streaming": + raise _to_http_error(AppError(code=ErrorCode.INVALID_WORKFLOW_TYPE, message="/api/workflows/stream 仅支持 response_mode=streaming", status_code=400)) stream_cfg = Config.get_section("stream") progress_interval = float(stream_cfg.get("progress_interval", 0.3)) @@ -147,16 +144,16 @@ def run_workflow_stream(payload: AgentInput, workflow_manager=Depends(get_workfl async def event_stream(): try: - slog.log("INFO", "stream.start", trace_id, {"workflow_type": payload.workflow_type}) + slog.log("INFO", "stream.start", trace_id, {"workflow_type": WorkflowType.CONVERSATION.value}) # 1) 先仅生成 SQL(不执行 SR API) result = await asyncio.to_thread( workflow_manager.execute_workflow, - workflow_type, - payload.input, - payload.session_id, + WorkflowType.CONVERSATION, + payload.query, + payload.conversation_id, skip_sr_api=True, ) - conversation_id = str(result.get("session_id") or payload.session_id or task_id) + conversation_id = str(result.get("session_id") or payload.conversation_id or task_id) context = (result.get("result") or {}).get("context") or {} sql_text = str(context.get("final_sql") or "") @@ -186,7 +183,7 @@ def run_workflow_stream(payload: AgentInput, workflow_manager=Depends(get_workfl yield "event: end\ndata: [DONE]\n\n" except Exception as e: slog.log("ERROR", "stream.failed", trace_id, error_code=ErrorCode.INTERNAL_ERROR.value, payload={"error": str(e)}) - conversation_id = str(payload.session_id or task_id) + conversation_id = str(payload.conversation_id or task_id) yield _build_message(conversation_id, str(e)) yield "event: end\ndata: [DONE]\n\n" diff --git a/schemas/agent_input.py b/schemas/agent_input.py index f80f456..a3b2a24 100644 --- a/schemas/agent_input.py +++ b/schemas/agent_input.py @@ -1,9 +1,16 @@ -from typing import Optional -from pydantic import BaseModel +from typing import Any, Dict, Optional + +from pydantic import BaseModel, ConfigDict, Field class AgentInput(BaseModel): - """Agent 输入模型""" - input: str - session_id: Optional[str] = None + """统一工作流输入模型(新版本 DTO)""" + + model_config = ConfigDict(extra="forbid") + + query: str + conversation_id: Optional[str] = None workflow_type: str = "conversation" + response_mode: str = "blocking" + user: Optional[str] = None + inputs: Dict[str, Any] = Field(default_factory=dict) diff --git a/schemas/chat_message_request.py b/schemas/chat_message_request.py new file mode 100644 index 0000000..bdb636a --- /dev/null +++ b/schemas/chat_message_request.py @@ -0,0 +1,22 @@ +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class ChatMessageFileDTO(BaseModel): + """文件输入模型(预留)""" + + model_config = ConfigDict(extra="allow") + + +class ChatMessageRequestDTO(BaseModel): + """/api/workflows/stream 请求模型(对齐标准 chat message 请求)""" + + model_config = ConfigDict(extra="forbid") + + query: str + inputs: Dict[str, Any] = Field(default_factory=dict) + response_mode: str = "streaming" + user: Optional[str] = None + conversation_id: Optional[str] = None + files: List[ChatMessageFileDTO] = Field(default_factory=list) diff --git a/schemas/stream_input.py b/schemas/stream_input.py new file mode 100644 index 0000000..af40fe9 --- /dev/null +++ b/schemas/stream_input.py @@ -0,0 +1,7 @@ +from schemas.agent_input import AgentInput + + +class StreamInputDTO(AgentInput): + """流式输入 DTO(与统一 AgentInput 保持一致)""" + + response_mode: str = "streaming" diff --git a/scripts/test_endpoints.py b/scripts/test_endpoints.py index 0dbed19..ad13399 100644 --- a/scripts/test_endpoints.py +++ b/scripts/test_endpoints.py @@ -85,23 +85,32 @@ def main() -> None: _get(client, f"{base}/nacos/status") elif choice == "3": payload = { - "input": "查询 SO 4020438779 的 eta 信息", - "session_id": None, + "query": "查询 SO 4020438779 的 eta 信息", + "conversation_id": None, "workflow_type": "conversation", + "response_mode": "blocking", + "user": "tester", + "inputs": {}, } _post(client, f"{base}/api/workflows", payload) elif choice == "4": payload = { - "input": "查询 SO 4016769041 的 eta 信息", - "session_id": None, - "workflow_type": "conversation", + "query": "查询 SO 4016769041 的 eta 信息", + "conversation_id": None, + "response_mode": "streaming", + "user": "tester", + "inputs": {}, + "files": [], } _stream_sse(client, f"{base}/api/workflows/stream", payload) elif choice == "5": payload = { - "input": "查询 SO 4020438779 的 eta 信息", - "session_id": None, + "query": "查询 SO 4020438779 的 eta 信息", + "conversation_id": None, "workflow_type": "conversation", + "response_mode": "blocking", + "user": "tester", + "inputs": {}, } _post(client, f"{base}/api/sql/generate", payload) elif choice == "6":