init

Jenkinsfile

@@ -1,3 +1,114 @@
+pipeline {
+    agent any
+
+    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
+                            echo "登录到镜像仓库 ${REGISTRY_URL}..."
+                            echo "${REGISTRY_PASS}" | docker login ${REGISTRY_URL} -u ${REGISTRY_USER} --password-stdin
+                        """
+                    }
+
+                    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') {
@@ -24,6 +135,11 @@ stage('Deploy to k3s') {
                                         done
                                     fi
 
+                                    if [ -z "\$KUBECTL_PATH" ]; then
+                                        echo "kubectl 未找到，请检查 Jenkins 节点环境"
+                                        exit 1
+                                    fi
+
                                     echo "使用kubectl路径: \$KUBECTL_PATH"
 
                                     # 定义kubectl函数
@@ -43,22 +159,97 @@ stage('Deploy to k3s') {
                                         --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
+                                    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（注意 deployment 名称是 more-dots）
+                                    # 确保使用imagePullSecret
-                            k patch deployment more-dots -n ${params.DEPLOY_ENV} \\
+                                    k patch deployment python-app -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 restart deployment/python-app -n ${params.DEPLOY_ENV} 2>/dev/null || true
-                            k rollout status deployment/more-dots -n ${params.DEPLOY_ENV} --timeout=300s
+                                    k rollout status deployment/python-app -n ${params.DEPLOY_ENV} --timeout=300s
                                 """
                             }
                         }
 
-                // cluster2 部分同样修改...
+                        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 "构建失败，请检查日志"
+        }
+    }
+}

README.md

@@ -125,9 +125,12 @@ manager_alt = WorkflowManager(default_model_section="gpt-3.5-turbo")
 
 ```json
 {
-    "input": "你好，帮我算 1 + 2",
+    "query": "你好，帮我算 1 + 2",
-    "session_id": null,
+    "conversation_id": null,
-    "workflow_type": "conversation"
+    "workflow_type": "conversation",
+    "response_mode": "blocking",
+    "user": "demo",
+    "inputs": {}
 }
 ```
 

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,
+            user_input=payload.query,
-            session_id=payload.session_id,
+            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,
+        user_input=payload.query,
-        session_id=payload.session_id,
+        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:
+    if payload.response_mode != "streaming":
-        raise _to_http_error(AppError(code=ErrorCode.INVALID_WORKFLOW_TYPE, message="仅支持对话工作流的流式输出", status_code=400))
+        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,
+                WorkflowType.CONVERSATION,
-                payload.input,
+                payload.query,
-                payload.session_id,
+                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"
 

schemas/agent_input.py

@@ -1,9 +1,16 @@
-from typing import Optional
+from typing import Any, Dict, Optional
-from pydantic import BaseModel
+
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class AgentInput(BaseModel):
-    """Agent 输入模型"""
+    """统一工作流输入模型（新版本 DTO）"""
-    input: str
+
-    session_id: Optional[str] = None
+    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)

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)

schemas/stream_input.py

@@ -0,0 +1,7 @@
+from schemas.agent_input import AgentInput
+
+
+class StreamInputDTO(AgentInput):
+    """流式输入 DTO（与统一 AgentInput 保持一致）"""
+
+    response_mode: str = "streaming"

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 信息",
+                    "query": "查询 SO 4020438779 的 eta 信息",
-                    "session_id": None,
+                    "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 信息",
+                    "query": "查询 SO 4016769041 的 eta 信息",
-                    "session_id": None,
+                    "conversation_id": None,
-                    "workflow_type": "conversation",
+                    "response_mode": "streaming",
+                    "user": "tester",
+                    "inputs": {},
+                    "files": [],
                 }
                 _stream_sse(client, f"{base}/api/workflows/stream", payload)
             elif choice == "5":
                 payload = {
-                    "input": "查询 SO 4020438779 的 eta 信息",
+                    "query": "查询 SO 4020438779 的 eta 信息",
-                    "session_id": None,
+                    "conversation_id": None,
                     "workflow_type": "conversation",
+                    "response_mode": "blocking",
+                    "user": "tester",
+                    "inputs": {},
                 }
                 _post(client, f"{base}/api/sql/generate", payload)
             elif choice == "6":