主题
Jenkins CI/CD Pipeline: Java + Python 并行构建 & AI 代码审查集成指南
版本: 1.0 | 日期: 2026-08-25 | Build: #12 (SUCCESS)
目录
- 1. 概述
- 2. 系统架构
- 3. 环境前提
- 4. 项目结构
- 5. 步骤一：创建 Java 基础项目
- 6. 步骤二：创建 Python 源代码
- 7. 步骤三：编写 Python 单元测试
- 8. 步骤四：编写 AI 代码审查脚本
- 9. 步骤五：编写 Jenkinsfile
- 10. 步骤六：推送到 GitLab
- 11. 步骤七：Jenkins 构建与验证
- 12. 问题排查与踩坑记录
- 13. AI 审查报告样例
- 14. 完整文件清单
1. 概述
本指南记录了如何将 Python 测试 和 Ollama AI 代码审查 集成到一个已有的 Java/Maven Jenkins CI/CD 流水线中。最终实现的效果：
GitLab Push → Webhook → Jenkins 触发构建
↓
并行构建:
├─ Java Agent (maven21) → mvn compile → mvn test (4/4 通过) → mvn package (.jar)
└─ Python Agent (python) → pylint → pytest (36/36 通过)
↓
AI Code Review (Python Agent):
1. 收集代码变更 (git diff / 源文件扫描)
2. 调用 Ollama Chat API (Qwen3.6:27B, think:false)
3. 生成结构化中文审查报告
4. 归档 ai-review-report.md + ai-review-result.json
↓
GitLab Commit Status 回调 (jenkins: success, ai-review: success)关键数据 (Build #12)：
- 总耗时：103 秒
- Java 测试：4/4 通过
- Python 测试：36/36 通过
- AI 审查耗时：62.6 秒
- 代码质量评分：7/10
2. 系统架构
┌─────────────────────────────────────────────────────────────────┐
│ Jenkins (K8s) │
│ ┌──────────────┐ ┌──────────────────────────────────┐ │
│ │ Jenkins Pod │────▶│ Kubernetes Cloud Plugin │ │
│ │ (jenkins-0) │ │ (动态 Agent Pod 创建/销毁) │ │
│ └──────────────┘ └────────┬──────────────┬──────────┘ │
│ │ │ │
│ ┌───────────▼──┐ ┌──────▼──────────┐ │
│ │ maven21 Pod │ │ python Pod │ │
│ │ (JDK+Maven) │ │ (Python Alpine) │ │
│ │ - git ✓ │ │ - git ✗ │ │
│ │ - curl ✓ │ │ - curl ✗ │ │
│ │ - bash ✓ │ │ - bash ✗ │ │
│ └──────────────┘ └─────────────────┘ │
│ ↓ Python urllib │
│ ┌───▼──────────────────┐ │
│ │ Ollama Server │ │
│ │ 192.168.122.9:11434 │ │
│ │ Model: qwen3.6:27b │ │
│ └──────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ GitLab │ │
│ │ Webhook → Jenkins Build API │ │
│ │ Commit Status API ← Jenkins Callback │ │
│ └──────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘Agent Pod 能力对比
| 工具 | maven21 Agent | python Agent | 影响 |
|---|---|---|---|
git | ✅ | ✅ | - |
curl | ✅ | ❌ | AI Review 需用 Python urllib 替代 |
bash | ✅ | ❌ | Shell 脚本需用 sh 而非 bash |
mvn | ✅ | ❌ | Java 构建只能在 maven agent 执行 |
python3 | ❌ | ✅ | Python 测试/审查只能在 python agent 执行 |
pip3 | ❌ | ✅ | - |
3. 环境前提
| 组件 | 版本/地址 | 说明 |
|---|---|---|
| Jenkins | K8s Pod (jenkins-0) | 运行在 namespace jenkins |
| GitLab | https://ai-ear.cn:8443 | 项目 ID: 4 |
| Ollama | http://192.168.122.9:11434 | 模型: qwen3.6:27b |
| Kubernetes | K8s Cloud Plugin | 动态 Agent Pod |
| Maven Agent Label | maven21 | 含 JDK + Maven + git + curl + bash |
| Python Agent Label | python | Alpine 镜像，含 Python 3.12，无 curl/bash/git |
Jenkins 凭证
| ID | 类型 | 用途 |
|---|---|---|
gitlab-git-user | Username/Password | GitLab Git 仓库克隆 |
Jenkins 插件
| 插件 | 用途 |
|---|---|
| GitLab Plugin | Commit Status 回调 |
| JUnit Plugin | 测试结果展示 |
| Kubernetes Plugin | 动态 Agent Pod |
4. 项目结构
demo-app-v2/
├── Jenkinsfile # 流水线定义
├── ai-code-review.sh # AI 代码审查脚本
├── pom.xml # Maven 项目配置
├── src/
│ ├── main/
│ │ ├── java/cn/ai_ear/demo/
│ │ │ └── App.java # Java 主类
│ │ └── python/
│ │ ├── calculator.py # 计算器模块
│ │ └── utils.py # 工具函数模块
│ └── test/
│ ├── java/cn/ai_ear/demo/
│ │ └── AppTest.java # Java 单元测试 (4个)
│ └── python/
│ ├── test_calculator.py # 计算器测试 (12个)
│ └── test_utils.py # 工具函数测试 (24个)
└── target/
└── demo-app-1.0.12.jar # 构建产物 (归档)5. 步骤一：创建 Java 基础项目
5.1 pom.xml
xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.ai-ear</groupId>
<artifactId>demo-app</artifactId>
<!-- BUILD_NUMBER 由 Jenkins 注入 -->
<version>1.0.${env.BUILD_NUMBER}</version>
<packaging>jar</packaging>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.2</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>5.2 App.java
java
package cn.ai_ear.demo;
public class App {
public static void main(String[] args) {
System.out.println("Hello Jenkins CI/CD!");
System.out.println("add(3, 5) = " + add(3, 5));
System.out.println("multiply(4, 6) = " + multiply(4, 6));
System.out.println("isPalindrome('racecar') = " + isPalindrome("racecar"));
System.out.println("fibonacci(10) = " + fibonacci(10));
}
public static int add(int a, int b) {
return a + b;
}
public static int multiply(int a, int b) {
return a * b;
}
public static boolean isPalindrome(String s) {
if (s == null) return false;
String cleaned = s.replaceAll("[^a-zA-Z0-9]", "").toLowerCase();
String reverse = new StringBuilder(cleaned).reverse().toString();
return cleaned.equals(reverse);
}
public static int fibonacci(int n) {
if (n <= 1) return n;
int a = 0, b = 1;
for (int i = 2; i <= n; i++) {
int temp = a + b;
a = b;
b = temp;
}
return b;
}
}5.3 AppTest.java
java
package cn.ai_ear.demo;
import org.junit.Test;
import static org.junit.Assert.*;
public class AppTest {
@Test
public void testAdd() {
assertEquals(8, App.add(3, 5));
assertEquals(0, App.add(-1, 1));
}
@Test
public void testMultiply() {
assertEquals(24, App.multiply(4, 6));
assertEquals(0, App.multiply(0, 5));
}
@Test
public void testIsPalindrome() {
assertTrue(App.isPalindrome("racecar"));
assertFalse(App.isPalindrome("hello"));
assertTrue(App.isPalindrome("A man a plan a canal Panama"));
}
@Test
public void testFibonacci() {
assertEquals(0, App.fibonacci(0));
assertEquals(1, App.fibonacci(1));
assertEquals(55, App.fibonacci(10));
}
}6. 步骤二：创建 Python 源代码
6.1 calculator.py
python
"""Calculator module - demonstrates basic arithmetic operations."""
class Calculator:
"""A simple calculator class supporting basic math operations."""
def __init__(self):
self.history = []
def add(self, a: float, b: float) -> float:
"""Add two numbers and record in history."""
result = a + b
self.history.append(f"{a} + {b} = {result}")
return result
def subtract(self, a: float, b: float) -> float:
"""Subtract b from a."""
result = a - b
self.history.append(f"{a} - {b} = {result}")
return result
def multiply(self, a: float, b: float) -> float:
"""Multiply two numbers."""
result = a * b
self.history.append(f"{a} * {b} = {result}")
return result
def divide(self, a: float, b: float) -> float:
"""Divide a by b. Raises ValueError on division by zero."""
if b == 0:
raise ValueError("Cannot divide by zero")
result = a / b
self.history.append(f"{a} / {b} = {result}")
return result
def power(self, base: float, exponent: float) -> float:
"""Raise base to the power of exponent."""
result = base ** exponent
self.history.append(f"{base} ** {exponent} = {result}")
return result
def modulo(self, a: float, b: float) -> float:
"""Return the remainder of a divided by b."""
if b == 0:
raise ValueError("Cannot modulo by zero")
result = a % b
self.history.append(f"{a} % {b} = {result}")
return result
def get_history(self) -> list:
"""Return the calculation history."""
return self.history.copy()
def clear_history(self):
"""Clear the calculation history."""
self.history.clear()
def main():
"""Demo usage of the Calculator class."""
calc = Calculator()
print(f"10 + 3 = {calc.add(10, 3)}")
print(f"10 - 3 = {calc.subtract(10, 3)}")
print(f"10 * 3 = {calc.multiply(10, 3)}")
print(f"10 / 3 = {calc.divide(10, 3):.4f}")
print(f"2 ** 8 = {calc.power(2, 8)}")
print(f"10 % 3 = {calc.modulo(10, 3)}")
print(f"\nHistory: {calc.get_history()}")
if __name__ == "__main__":
main()6.2 utils.py
python
"""Utility functions for data processing and string manipulation."""
import re
from collections import Counter
from typing import Any
def is_palindrome(text: str) -> bool:
"""Check if text is a palindrome (case-insensitive, ignoring non-alphanumeric)."""
cleaned = re.sub(r'[^a-zA-Z0-9]', '', text).lower()
return cleaned == cleaned[::-1]
def word_frequency(text: str) -> dict[str, int]:
"""Return a dictionary of word frequencies in the text."""
words = re.findall(r'\b\w+\b', text.lower())
return dict(Counter(words))
def flatten(nested_list: list) -> list:
"""Flatten a nested list into a single-level list."""
result = []
for item in nested_list:
if isinstance(item, list):
result.extend(flatten(item))
else:
result.append(item)
return result
def camel_to_snake(name: str) -> str:
"""Convert camelCase string to snake_case."""
s1 = re.sub(r'(.)([A-Z][a-z]+)', r'\1_\2', name)
return re.sub(r'([a-z0-9])([A-Z])', r'\1_\2', s1).lower()
def snake_to_camel(name: str) -> str:
"""Convert snake_case string to camelCase."""
components = name.split('_')
return components[0] + ''.join(x.title() for x in components[1:])
def safe_divide(a: float, b: float, default: float = 0.0) -> float:
"""Safely divide a by b, returning default if b is zero."""
try:
return a / b
except (ZeroDivisionError, TypeError):
return default
def chunk_list(lst: list, chunk_size: int) -> list[list]:
"""Split a list into chunks of the specified size."""
if chunk_size <= 0:
raise ValueError("chunk_size must be positive")
return [lst[i:i + chunk_size] for i in range(0, len(lst), chunk_size)]
def deep_get(data: dict, path: str, default: Any = None) -> Any:
"""Safely access nested dictionary values using dot-separated path."""
keys = path.split('.')
current = data
for key in keys:
if isinstance(current, dict) and key in current:
current = current[key]
else:
return default
return current
def main():
"""Demo usage."""
print(f"'racecar' is palindrome: {is_palindrome('racecar')}")
print(f"'hello' is palindrome: {is_palindrome('hello')}")
print(f"Word freq: {word_frequency('the cat sat on the mat the cat')}")
print(f"Flatten: {flatten([1, [2, 3], [4, [5, 6]]])}")
print(f"camelCase → snake_case: {camel_to_snake('myVariableName')}")
print(f"snake_case → camelCase: {snake_to_camel('my_variable_name')}")
print(f"Safe divide 10/3: {safe_divide(10, 3):.4f}")
print(f"Safe divide 10/0: {safe_divide(10, 0)}")
print(f"Chunk [1..10] by 3: {chunk_list(list(range(1, 11)), 3)}")
print(f"Deep get: {deep_get({'a': {'b': {'c': 42}}}, 'a.b.c')}")
if __name__ == "__main__":
main()说明：Python 源文件放在
src/main/python/目录下，与 Java 的src/main/java/保持一致的 Maven 目录约定。
7. 步骤三：编写 Python 单元测试
7.1 test_calculator.py (12 个测试)
python
"""Unit tests for the Calculator class."""
import sys
import os
import unittest
# Add the source directory to the path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'main', 'python'))
from calculator import Calculator
class TestCalculator(unittest.TestCase):
"""Test suite for the Calculator class."""
def setUp(self):
"""Set up a fresh Calculator instance for each test."""
self.calc = Calculator()
def test_add(self):
self.assertEqual(self.calc.add(2, 3), 5)
self.assertEqual(self.calc.add(-1, 1), 0)
self.assertEqual(self.calc.add(-1, -2), -3)
self.assertAlmostEqual(self.calc.add(0.1, 0.2), 0.3, places=10)
def test_subtract(self):
self.assertEqual(self.calc.subtract(10, 3), 7)
self.assertEqual(self.calc.subtract(3, 10), -7)
self.assertEqual(self.calc.subtract(0, 0), 0)
def test_multiply(self):
self.assertEqual(self.calc.multiply(2, 3), 6)
self.assertEqual(self.calc.multiply(0, 5), 0)
self.assertEqual(self.calc.multiply(-2, -3), 6)
self.assertEqual(self.calc.multiply(-2, 3), -6)
def test_divide(self):
self.assertEqual(self.calc.divide(10, 2), 5)
self.assertAlmostEqual(self.calc.divide(10, 3), 3.3333, places=3)
self.assertEqual(self.calc.divide(0, 5), 0)
def test_divide_by_zero(self):
with self.assertRaises(ValueError):
self.calc.divide(10, 0)
def test_power(self):
self.assertEqual(self.calc.power(2, 8), 256)
self.assertEqual(self.calc.power(5, 0), 1)
self.assertEqual(self.calc.power(9, 0.5), 3)
def test_modulo(self):
self.assertEqual(self.calc.modulo(10, 3), 1)
self.assertEqual(self.calc.modulo(15, 5), 0)
def test_modulo_by_zero(self):
with self.assertRaises(ValueError):
self.calc.modulo(10, 0)
def test_history(self):
self.calc.add(1, 2)
self.calc.multiply(3, 4)
history = self.calc.get_history()
self.assertEqual(len(history), 2)
self.assertIn("1 + 2 = 3", history[0])
self.assertIn("3 * 4 = 12", history[1])
def test_clear_history(self):
self.calc.add(1, 2)
self.calc.clear_history()
self.assertEqual(len(self.calc.get_history()), 0)
if __name__ == '__main__':
unittest.main()7.2 test_utils.py (24 个测试)
python
"""Unit tests for utility functions."""
import sys
import os
import unittest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'main', 'python'))
from utils import (
is_palindrome, word_frequency, flatten, camel_to_snake,
snake_to_camel, safe_divide, chunk_list, deep_get
)
class TestIsPalindrome(unittest.TestCase):
def test_simple_palindromes(self):
self.assertTrue(is_palindrome("racecar"))
self.assertTrue(is_palindrome("madam"))
self.assertTrue(is_palindrome("level"))
def test_non_palindromes(self):
self.assertFalse(is_palindrome("hello"))
self.assertFalse(is_palindrome("world"))
def test_case_insensitive(self):
self.assertTrue(is_palindrome("Racecar"))
self.assertTrue(is_palindrome("A man a plan a canal Panama"))
def test_with_special_chars(self):
self.assertTrue(is_palindrome("Was it a car or a cat I saw?"))
def test_empty_and_single(self):
self.assertTrue(is_palindrome(""))
self.assertTrue(is_palindrome("a"))
class TestWordFrequency(unittest.TestCase):
def test_basic(self):
result = word_frequency("the cat sat on the mat")
self.assertEqual(result["the"], 2)
self.assertEqual(result["cat"], 1)
def test_case_insensitive(self):
result = word_frequency("Hello hello HELLO")
self.assertEqual(result["hello"], 3)
def test_empty(self):
result = word_frequency("")
self.assertEqual(result, {})
class TestFlatten(unittest.TestCase):
def test_simple(self):
self.assertEqual(flatten([1, [2, 3], 4]), [1, 2, 3, 4])
def test_nested(self):
self.assertEqual(flatten([1, [2, [3, [4]]]]), [1, 2, 3, 4])
def test_empty(self):
self.assertEqual(flatten([]), [])
self.assertEqual(flatten([[], [[]]]), [])
def test_already_flat(self):
self.assertEqual(flatten([1, 2, 3]), [1, 2, 3])
class TestCamelToSnake(unittest.TestCase):
def test_basic(self):
self.assertEqual(camel_to_snake("camelCase"), "camel_case")
self.assertEqual(camel_to_snake("myVariableName"), "my_variable_name")
def test_single_word(self):
self.assertEqual(camel_to_snake("hello"), "hello")
def test_acronyms(self):
self.assertEqual(camel_to_snake("HTTPResponse"), "http_response")
class TestSnakeToCamel(unittest.TestCase):
def test_basic(self):
self.assertEqual(snake_to_camel("snake_case"), "snakeCase")
self.assertEqual(snake_to_camel("my_variable_name"), "myVariableName")
def test_single_word(self):
self.assertEqual(snake_to_camel("hello"), "hello")
class TestSafeDivide(unittest.TestCase):
def test_normal(self):
self.assertEqual(safe_divide(10, 2), 5.0)
def test_zero(self):
self.assertEqual(safe_divide(10, 0), 0.0)
self.assertEqual(safe_divide(10, 0, -1.0), -1.0)
def test_type_error(self):
self.assertEqual(safe_divide("a", "b"), 0.0)
class TestChunkList(unittest.TestCase):
def test_basic(self):
self.assertEqual(chunk_list([1, 2, 3, 4, 5], 2), [[1, 2], [3, 4], [5]])
def test_exact_chunks(self):
self.assertEqual(chunk_list([1, 2, 3, 4], 2), [[1, 2], [3, 4]])
def test_invalid_size(self):
with self.assertRaises(ValueError):
chunk_list([1, 2], 0)
class TestDeepGet(unittest.TestCase):
def test_basic(self):
data = {"a": {"b": {"c": 42}}}
self.assertEqual(deep_get(data, "a.b.c"), 42)
def test_missing_key(self):
data = {"a": {"b": 1}}
self.assertIsNone(deep_get(data, "a.c"))
self.assertEqual(deep_get(data, "a.c", "default"), "default")
def test_top_level(self):
data = {"key": "value"}
self.assertEqual(deep_get(data, "key"), "value")
if __name__ == '__main__':
unittest.main()7.3 关键注意事项
⚠️
sys.path导入路径：由于没有使用setup.py/pyproject.toml进行正式打包，测试文件需要手动将源码目录加入 Python 路径：pythonsys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'main', 'python'))
⚠️ 测试断言修正：
camel_to_snake("HTTPResponse")的正确结果是"http_response"（连续大写字母被识别为一个缩写词），而非"h_t_t_p_response"。
8. 步骤四：编写 AI 代码审查脚本
8.1 设计思路
由于 Python Alpine Agent 镜像中 没有 curl、没有 bash，因此 AI 审查脚本的设计需要：
- 使用
#!/bin/sh而非#!/bin/bash - 使用
set -eu而非set -euo pipefail（Alpine sh 不支持pipefail） - 所有 HTTP 请求通过 Python
urllib实现，不依赖curl - 代码收集通过 Python
glob+subprocess实现，不强依赖git diff
8.2 ai-code-review.sh 完整代码
bash
#!/bin/sh
# =============================================================================
# AI Code Review Script - Ollama Chat API (Qwen3.6:27B, think:false)
# Pure Python implementation - no curl/git dependencies required
# =============================================================================
set -eu
export OLLAMA_URL="${OLLAMA_URL:-http://192.168.122.9:11434}"
export OLLAMA_MODEL="${OLLAMA_MODEL:-qwen3.6:27b}"
echo "========================================"
echo " AI Code Review (${OLLAMA_MODEL})"
echo "========================================"
python3 << 'AIREVIEW'
import json, os, urllib.request, time, subprocess, glob
OLLAMA_URL = os.environ.get('OLLAMA_URL', 'http://192.168.122.9:11434')
OLLAMA_MODEL = os.environ.get('OLLAMA_MODEL', 'qwen3.6:27b')
MAX_DIFF_CHARS = 6000
# ─── 1. 收集代码变更 ───────────────────────────────────────────────
diff = ""
# 策略 A: 尝试 git diff（如果 git 可用）
try:
for base in ['main', 'master']:
try:
result = subprocess.run(
['git', 'diff', f'origin/{base}', '--',
'*.java', '*.py', '*.xml', 'Jenkinsfile'],
capture_output=True, text=True, timeout=10
)
if result.stdout.strip():
diff = result.stdout
print(f" ✓ Git diff vs origin/{base}")
break
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
if not diff:
try:
result = subprocess.run(
['git', 'diff', 'HEAD~1', '--',
'*.java', '*.py', '*.xml', 'Jenkinsfile'],
capture_output=True, text=True, timeout=10
)
if result.stdout.strip():
diff = result.stdout
print(" ✓ Git diff from last commit")
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
except Exception:
pass
# 策略 B: 回退到直接读取源文件（git 不可用时）
if not diff:
print(" ⚠ No git diff, reviewing source files directly...")
for pattern in ['src/main/**/*.java', 'src/main/**/*.py']:
for f in glob.glob(pattern, recursive=True):
if 'test' not in f.lower():
try:
with open(f) as fh:
content = fh.read()[:2000]
diff += f"\n=== {f} ===\n{content}\n"
except Exception:
pass
# 无代码可审查，跳过
if not diff:
with open('ai-review-result.json', 'w') as f:
json.dump({"status": "skipped", "reason": "no code found"}, f)
print(" ⚠ No code to review")
exit(0)
# ─── 2. 截断 & 统计 ────────────────────────────────────────────────
if len(diff) > MAX_DIFF_CHARS:
diff = diff[:MAX_DIFF_CHARS] + "\n... [truncated]"
print(f" Code size: {len(diff)} chars")
java_files = len(glob.glob('src/main/**/*.java', recursive=True))
python_files = len(glob.glob('src/main/**/*.py', recursive=True))
test_java = len(glob.glob('src/test/**/*.java', recursive=True))
test_python = len(glob.glob('src/test/**/*.py', recursive=True))
print(f" Java: {java_files} src, {test_java} test | "
f"Python: {python_files} src, {test_python} test")
# ─── 3. 构建 Prompt ────────────────────────────────────────────────
prompt = f"""You are a senior code reviewer. Review the following code.
Files: {java_files} Java, {python_files} Python source files.
Code:
{diff}
Provide your review in this EXACT format (in Chinese):
## 📊 代码审查总结
- 代码质量评分: X/10
- 总体评价: (1-2句话)
## ✅ 优点
1. ...
2. ...
## ⚠️ 改进建议
1. ...
2. ...
## 🐛 潜在问题
1. ... (如有)
Focus on: correctness, security, performance, readability."""
# ─── 4. 调用 Ollama Chat API ───────────────────────────────────────
#
# 关键参数:
# "think": false → 禁用 Qwen3 的 think_token 推理机制，避免超时
# "num_predict": 800 → 限制输出 token 数
# "temperature": 0.3 → 较低温度保证输出稳定性
#
payload = json.dumps({
"model": OLLAMA_MODEL,
"messages": [{"role": "user", "content": prompt}],
"stream": False,
"think": False,
"options": {"num_predict": 800, "temperature": 0.3}
}).encode('utf-8')
print(f"\n Calling Ollama ({OLLAMA_MODEL})...")
start = time.time()
try:
req = urllib.request.Request(
f"{OLLAMA_URL}/api/chat",
data=payload,
headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req, timeout=180) as resp:
data = json.loads(resp.read().decode())
duration = round(time.time() - start, 1)
review_text = data.get('message', {}).get('content', 'No response')
# ─── 5. 保存结果 ──────────────────────────────────────────────
with open('ai-review-report.md', 'w') as f:
f.write(review_text)
with open('ai-review-result.json', 'w') as f:
json.dump({
"status": "success",
"model": OLLAMA_MODEL,
"duration_seconds": duration,
"java_files": java_files,
"python_files": python_files
}, f, indent=2)
print(f" ✓ AI review completed in {duration}s")
print()
print("=" * 50)
print(" AI Code Review Report")
print("=" * 50)
print()
print(review_text)
print()
print("=" * 50)
print(f" Done in {duration}s | ai-review-report.md")
print("=" * 50)
except Exception as e:
duration = round(time.time() - start, 1)
print(f" ✗ AI review failed after {duration}s: {e}")
with open('ai-review-result.json', 'w') as f:
json.dump({"status": "error", "reason": str(e)}, f)
with open('ai-review-report.md', 'w') as f:
f.write(f"AI review failed: {e}")
AIREVIEW
echo "========================================"
echo " AI Code Review Complete"
echo "========================================"
cat ai-review-result.json 2>/dev/null || true8.3 Ollama API 参数说明
| 参数 | 值 | 说明 |
|---|---|---|
model | qwen3.6:27b | Ollama 本地部署的大语言模型 |
stream | false | 非流式响应，一次性返回完整结果 |
think | false | 关键参数：禁用 think_token 机制 |
num_predict | 800 | 限制最大输出 token 数 |
temperature | 0.3 | 低温度 = 更确定性的输出 |
timeout | 180s | HTTP 请求超时时间 |
⚠️
"think": false是必须的。Qwen3.6:27B 默认启用 think_token 推理机制，会先生成大量内部推理内容再输出最终结果，导致响应极慢甚至超时。
9. 步骤五：编写 Jenkinsfile
9.1 完整 Jenkinsfile
groovy
pipeline {
agent { label 'maven21' }
options {
timestamps()
timeout(time: 20, unit: 'MINUTES')
buildDiscarder(logRotator(numToKeepStr: '10'))
}
environment {
OLLAMA_URL = 'http://192.168.122.9:11434'
OLLAMA_MODEL = 'qwen3.6:27b'
PYTHON_DIR = 'src/main/python'
PYTHON_TEST_DIR = 'src/test/python'
}
stages {
// ═══ 阶段1: Checkout ═══
stage('Checkout') {
steps {
container('maven') {
checkout scm
sh '''
echo "============================================"
echo " Build Info"
echo "============================================"
echo "Branch: ${GIT_BRANCH}"
echo "Commit: ${GIT_COMMIT}"
echo "Build#: ${BUILD_NUMBER}"
echo "============================================"
echo "Source files:"
find . -type f \\( -name "*.java" -o -name "*.py" \\) | sort
echo "============================================"
'''
}
}
}
// ═══ 阶段2: 并行构建 + 测试 ═══
stage('Build & Test') {
parallel {
stage('Java: Build + Test') {
steps {
container('maven') {
sh 'mvn -B -Dmaven.repo.local=$WORKSPACE/.m2 clean compile'
sh 'mvn -B -Dmaven.repo.local=$WORKSPACE/.m2 test'
}
}
post {
always {
junit allowEmptyResults: true,
testResults: 'target/surefire-reports/*.xml'
}
}
}
stage('Python: Lint + Test') {
agent { label 'python' }
steps {
container('python') {
sh '''
echo "=== Python Environment ==="
python3 --version
pip3 --version
echo ""
echo "=== Installing test dependencies ==="
pip3 install pytest pylint --quiet 2>&1
echo ""
echo "=== Pylint: Code Quality Check ==="
pylint --disable=C0114,C0115,C0116,R0903 \\
${PYTHON_DIR}/calculator.py \\
${PYTHON_DIR}/utils.py \\
--output-format=text \\
--score=yes 2>&1 || true
echo ""
echo "=== Pytest: Unit Tests ==="
python3 -m pytest ${PYTHON_TEST_DIR}/ \\
-v --tb=short \\
--junitxml=python-test-results.xml 2>&1
'''
}
}
post {
always {
junit allowEmptyResults: true,
testResults: 'python-test-results.xml'
}
}
}
}
}
// ═══ 阶段3: Java 打包 ═══
stage('Java: Package') {
steps {
container('maven') {
sh 'mvn -B -Dmaven.repo.local=$WORKSPACE/.m2 package -DskipTests'
}
}
post {
success {
archiveArtifacts artifacts: 'target/*.jar', fingerprint: true
}
}
}
// ═══ 阶段4: AI 代码审查 ═══
stage('AI Code Review') {
agent { label 'python' }
steps {
container('python') {
sh '''
echo "=== AI Code Review Environment ==="
python3 --version
which curl 2>/dev/null || \\
echo "curl: not available (using Python urllib)"
echo ""
echo "=== Running AI Code Review ==="
chmod +x ai-code-review.sh
sh ai-code-review.sh
'''
}
}
post {
always {
archiveArtifacts allowEmptyArchive: true,
artifacts: 'ai-review-report.md, ai-review-result.json'
}
}
}
}
// ═══ 全局后置动作 ═══
post {
success {
updateGitlabCommitStatus name: 'jenkins', state: 'success'
updateGitlabCommitStatus name: 'ai-review', state: 'success'
}
failure {
updateGitlabCommitStatus name: 'jenkins', state: 'failed'
}
}
}9.2 关键设计决策
| 决策 | 原因 |
|---|---|
默认 Agent 为 maven21 | Checkout 和 Java 构建阶段需要 Maven 环境 |
Python 阶段使用独立 agent { label 'python' } | 隔离到 Python Alpine Agent Pod |
AI Review 使用 python agent | 脚本纯 Python 实现，无需 Maven 环境 |
sh 而非 bash 执行脚本 | Alpine 无 bash |
| Pylint ` | |
allowEmptyArchive: true | AI Review 失败时不阻塞 artifact 归档 |
| `which curl 2>/dev/null |
9.3 流水线阶段流程图
┌─────────────┐
│ Checkout │ ← maven21 agent
└──────┬──────┘
│
┌──────▼──────────────────────────┐
│ Build & Test (parallel) │
│ ┌─────────────┐ ┌────────────┐ │
│ │Java: Build │ │Python: Lint│ │
│ │+ Test │ │+ Test │ │
│ │(maven21) │ │(python) │ │
│ │4/4 pass │ │36/36 pass │ │
│ └─────────────┘ └────────────┘ │
└──────┬──────────────────────────┘
│
┌──────▼─────────────┐
│ Java: Package │ ← maven21 agent
│ (.jar artifact) │
└──────┬─────────────┘
│
┌──────▼─────────────┐
│ AI Code Review │ ← python agent
│ (Ollama API) │
│ → report.md │
│ → result.json │
└──────┬─────────────┘
│
┌──────▼─────────────┐
│ Post Actions │
│ GitLab Callback │
└────────────────────┘10. 步骤六：推送到 GitLab
10.1 通过 GitLab API 提交文件
由于 Jenkins 环境通过 K8s 运行，文件提交通过 GitLab REST API 完成：
bash
# 环境变量
export TOKEN="xxx"
export GITLAB="https://ai-ear.cn:8443"
export PROJECT_ID=4
# 提交单个文件示例
curl -sk -X POST \
-H "PRIVATE-TOKEN: $TOKEN" \
-H "Content-Type: application/json" \
"$GITLAB/api/v4/projects/$PROJECT_ID/repository/commits" \
-d '{
"branch": "main",
"commit_message": "feat: add Python source and test files",
"actions": [
{
"action": "create",
"file_path": "src/main/python/calculator.py",
"content": "... (文件内容) ..."
}
]
}'10.2 批量提交多个文件 (Python 脚本)
python
import requests, json, urllib3
urllib3.disable_warnings()
HEADERS = {
'PRIVATE-TOKEN': 'glpat-xxx',
'Content-Type': 'application/json'
}
GITLAB = "https://ai-ear.cn:8443"
# 读取所有文件
files_to_commit = {
'Jenkinsfile': '/path/to/Jenkinsfile',
'ai-code-review.sh': '/path/to/ai-code-review.sh',
'src/main/python/calculator.py': '/path/to/calculator.py',
'src/main/python/utils.py': '/path/to/utils.py',
'src/test/python/test_calculator.py': '/path/to/test_calculator.py',
'src/test/python/test_utils.py': '/path/to/test_utils.py',
}
actions = []
for gitlab_path, local_path in files_to_commit.items():
with open(local_path) as f:
content = f.read()
actions.append({
'action': 'update', # 或 'create' 用于新文件
'file_path': gitlab_path,
'content': content
})
resp = requests.post(
f"{GITLAB}/api/v4/projects/4/repository/commits",
headers=HEADERS,
json={
'branch': 'main',
'commit_message': 'feat: integrate Python testing and AI code review',
'actions': actions
},
verify=False
)
print(json.dumps(resp.json(), indent=2))10.3 Webhook 自动触发
GitLab 已配置 Webhook，每次 push 到 main 分支会自动调用 Jenkins Build API 触发构建。无需手动触发。
11. 步骤七：Jenkins 构建与验证
11.1 手动触发构建
bash
export KUBECONFIG=~/.kube/config-122.31
kubectl exec -n jenkins jenkins-0 -c jenkins -- sh -c \
"curl -s -u 'admin:xxx' \
'http://localhost:8080/job/dev/job/gitlab-cicd-demo/build'"11.2 查看构建状态
bash
kubectl exec -n jenkins jenkins-0 -c jenkins -- sh -c \
"curl -s -u 'admin:xxx' \
'http://localhost:8080/job/dev/job/gitlab-cicd-demo/12/api/json'" \
| python3 -c "
import json,sys
d=json.load(sys.stdin)
print(f'Result={d.get(\"result\",\"IN_PROGRESS\")}')
print(f'Duration={d.get(\"duration\",0)/1000:.0f}s')
print(f'Building={d.get(\"building\")}')
for a in d.get('artifacts', []):
print(f' Artifact: {a.get(\"fileName\")}')
"11.3 查看控制台日志
bash
kubectl exec -n jenkins jenkins-0 -c jenkins -- sh -c \
"curl -s -u 'admin:xxx' \
'http://localhost:8080/job/dev/job/gitlab-cicd-demo/12/consoleText'"11.4 验证 GitLab Commit Status
bash
curl -sk -H "PRIVATE-TOKEN: $TOKEN" \
"$GITLAB/api/v4/projects/4/repository/commits/COMMIT_SHA/statuses" \
| python3 -c "
import json, sys
for s in json.load(sys.stdin):
print(f'{s[\"name\"]:15s} | {s[\"status\"]:8s} | {s.get(\"description\",\"-\")}')
"
# 预期输出:
# jenkins | success | success
# ai-review | success | success11.5 验证 Artifacts
Build #12 成功生成 3 个 artifact：
| Artifact | 路径 | 说明 |
|---|---|---|
demo-app-1.0.12.jar | target/ | Java 构建产物 |
ai-review-report.md | 根目录 | AI 代码审查 Markdown 报告 |
ai-review-result.json | 根目录 | AI 审查结果元数据 |
11.6 预期构建结果
✅ Checkout - 成功
✅ Java: Build + Test - 4/4 JUnit 通过
✅ Python: Lint + Test - 36/36 Pytest 通过, Pylint 执行
✅ Java: Package - demo-app-1.0.N.jar 打包
✅ AI Code Review - Ollama 审查完成, report.md 归档
✅ Post Actions - GitLab status: jenkins=success, ai-review=success12. 问题排查与踩坑记录
🔴 问题1：bash: not found (Build #11)
现象：
line 9: bash: not found
ERROR: script returned exit code 127原因：Python Alpine Agent 镜像中没有 bash。Alpine Linux 默认只包含 ash (BusyBox sh)。
解决：
diff
# Jenkinsfile
- bash ai-code-review.sh
+ sh ai-code-review.sh
# ai-code-review.sh
- #!/bin/bash
- set -euo pipefail
+ #!/bin/sh
+ set -eu教训：Alpine 镜像中永远不要假设
bash存在。始终使用#!/bin/sh并避免pipefail等 bash-only 特性。
🔴 问题2：which curl 导致 exit code 1 (Build #10)
现象：
+ which curl
ERROR: script returned exit code 1原因：which curl 在 Alpine 中找不到 curl 二进制文件，返回非零退出码。由于脚本启用了 set -e，非零退出码导致脚本立即终止。
解决：
diff
- which curl
+ which curl 2>/dev/null || echo "curl: not available (using Python urllib)"教训：在
set -e环境下，任何可能失败的命令都需要用|| true或|| echo ...保护。
🔴 问题3：test_acronyms 断言失败 (Build #8)
现象：
FAILED src/test/python/test_utils.py::TestCamelToSnake::test_acronyms
AssertionError: 'http_response' != 'h_t_t_p_response'原因：camel_to_snake() 函数的实现逻辑会将连续大写字母（如 HTTP）识别为一个缩写词整体，因此 HTTPResponse 正确转换为 http_response。但最初测试断言写的是 h_t_t_p_response。
解决：
diff
# test_utils.py
- self.assertEqual(camel_to_snake("HTTPResponse"), "h_t_t_p_response")
+ self.assertEqual(camel_to_snake("HTTPResponse"), "http_response")教训：编写测试前应先验证被测函数的实际行为，避免基于假设编写断言。
🟡 问题4：Ollama Qwen3.6 超时
现象：AI Review 请求发出后长时间无响应，最终超时。
原因：Qwen3.6:27B 默认启用 think_token 推理机制（CoT 思维链），模型会先生成大量内部推理 token（可能数千 token），然后再输出最终结果。27B 参数的模型生成这些 token 需要较长时间。
解决：在 Ollama Chat API payload 中添加 "think": false：
json
{
"model": "qwen3.6:27b",
"messages": [{"role": "user", "content": "..."}],
"stream": false,
"think": false,
"options": {
"num_predict": 800,
"temperature": 0.3
}
}教训：使用具有推理能力的模型（如 Qwen3、DeepSeek-R1 等）时，务必检查是否需要显式关闭 think_token。
🟡 问题5：Alpine Agent 能力受限
现象：Python Agent 镜像是 Alpine-based，缺少大量常用工具。
能力对比：
| 工具 | maven21 | python (Alpine) | 替代方案 |
|---|---|---|---|
git | ✅ | ✅ | - |
curl | ✅ | ❌ | Python urllib |
bash | ✅ | ❌ | sh |
wget | ❓ | ❌ | Python urllib |
jq | ❓ | ❌ | Python json |
影响：所有需要在 Python Agent 上执行的外部交互（如 Ollama API 调用）必须使用 Python 标准库实现，不能依赖 shell 工具。
架构建议：如果需要频繁使用 AI 审查功能，考虑构建一个包含更多工具的自定义 Agent 镜像。
13. AI 审查报告样例
以下是 Build #12 中 Qwen3.6:27B 生成的代码审查报告：
markdown
## 📊 代码审查总结
- 代码质量评分: 7/10
- 总体评价: 代码结构清晰，类型注解和文档字符串规范，但存在明显的截断错误
（Python文件未完整），且部分逻辑缺乏边界条件处理或存在冗余。
## ✅ 优点
1. **良好的文档习惯**：Python 文件使用了标准的 Docstrings，Java 方法命名清晰，
便于理解意图。
2. **类型安全**：Python 代码广泛使用了类型提示（Type Hints），提高了代码的
可维护性和 IDE 支持体验。
3. **防御性编程意识**：`Calculator` 类中的除法和取模操作检查了零值，
`safe_divide` 使用了 try-except 块，体现了对异常处理的关注。
4. **不可变性保护**：`Calculator.get_history()` 返回副本而非引用，防止外部
意外修改内部状态。
## ⚠️ 改进建议
1. **Java `isPalindrome` 性能优化**：当前实现创建了两个新字符串对象。对于
长字符串，建议使用双指针法在原字符串上进行比较，以减少内存分配和 GC 压力。
2. **Python `flatten` 递归深度风险**：`flatten` 函数使用递归实现，对于极深
的嵌套列表可能导致 `RecursionError`。建议考虑使用迭代方式（如栈）或
生成器来处理大规模数据。
3. **Java 代码职责分离**：`App.java` 混合了入口点、工具方法和业务逻辑。
建议将工具方法移至独立的 `Utils` 类，保持 `App` 类仅作为启动入口。
4. **Python `word_frequency` 正则优化**：`\b\w+\b` 会包含数字和下划线。
如果仅针对英文单词，建议使用 `[a-zA-Z]+` 或根据具体需求调整正则。
## 🐛 潜在问题
1. **代码截断严重**：提供的 `calculator.py` 在 `main` 函数中截断，`utils.py`
在 `deep_get` 函数中截断。这导致无法评估完整逻辑的正确性。
2. **Java `isPalindrome` 空指针风险**：虽然检查了 `s == null`，但
`replaceAll` 的正则表达式在某些极端 Unicode 字符下可能行为不符合预期。
3. **Python `chunk_list` 类型提示**：返回类型 `list[list]` 在 Python 3.9+
中可用，但在更早版本中需使用 `typing.List[List]`。建议确认项目最低
Python 版本要求。14. 完整文件清单
| 文件 | 用途 | 所在阶段 |
|---|---|---|
Jenkinsfile | 流水线定义 | 全局 |
ai-code-review.sh | AI 代码审查脚本 | AI Code Review |
pom.xml | Maven 项目配置 | Java Build |
src/main/java/cn/ai_ear/demo/App.java | Java 主类 | Java Build + Test |
src/test/java/cn/ai_ear/demo/AppTest.java | Java 单元测试 (4个) | Java Build + Test |
src/main/python/calculator.py | 计算器模块 | Python Lint + Test |
src/main/python/utils.py | 工具函数模块 | Python Lint + Test |
src/test/python/test_calculator.py | 计算器测试 (12个) | Python Lint + Test |
src/test/python/test_utils.py | 工具函数测试 (24个) | Python Lint + Test |
构建产物
| Artifact | 说明 |
|---|---|
target/demo-app-1.0.{BUILD_NUMBER}.jar | Java JAR 包 |
ai-review-report.md | AI 代码审查 Markdown 报告 |
ai-review-result.json | AI 审查元数据 (status/model/duration) |
target/surefire-reports/*.xml | JUnit 测试报告 |
python-test-results.xml | Pytest 测试报告 |
附录 A: 调试命令速查
bash
# 设置 K8s 环境
export KUBECONFIG=~/.kube/config-122.31
# 查看所有 Agent Pods
kubectl get pods -n jenkins
# 查看特定构建的控制台日志
kubectl exec -n jenkins jenkins-0 -c jenkins -- sh -c \
"curl -s -u 'admin:xxx' \
'http://localhost:8080/job/dev/job/gitlab-cicd-demo/BUILD_NUM/consoleText'"
# 查看最近构建列表
kubectl exec -n jenkins jenkins-0 -c jenkins -- sh -c \
"curl -s -u 'admin:xxx' \
'http://localhost:8080/job/dev/job/gitlab-cicd-demo/api/json?tree=builds[number,result,building]'"
# 从 Python Agent Pod 内测试 Ollama 连通性
kubectl exec -it PYTHON_POD -n jenkins -c python -- python3 -c "
import urllib.request, json
req = urllib.request.Request('http://192.168.122.9:11434/api/tags')
with urllib.request.urlopen(req, timeout=10) as r:
data = json.loads(r.read())
for m in data.get('models', []):
print(f' {m[\"name\"]} ({m.get(\"size\",0)/1e9:.1f}GB)')
"
# 查看 GitLab 上项目的文件树
curl -sk -H "PRIVATE-TOKEN: $TOKEN" \
"$GITLAB/api/v4/projects/4/repository/tree?recursive=true&per_page=100"
# 查看 GitLab 上的 commit 历史
curl -sk -H "PRIVATE-TOKEN: $TOKEN" \
"$GITLAB/api/v4/projects/4/repository/commits?per_page=10" \
| python3 -c "
import json, sys
for c in json.load(sys.stdin):
print(f'{c[\"short_id\"]} {c[\"title\"]}')
"附录 B: Git Commit 历史
8605af8d fix: use sh instead of bash for Alpine compatibility
7f13a6ad fix: correct test_acronyms assertion (HTTPResponse -> http_response)
... feat: add AI code review script with Ollama integration
... feat: add Python source files and unit tests
... feat: initial Java Maven project with Jenkins pipeline附录 C: 测试覆盖汇总
Java 测试 (4 个, JUnit 4)
| 测试方法 | 断言内容 |
|---|---|
testAdd | add(3,5)=8, add(-1,1)=0 |
testMultiply | multiply(4,6)=24, multiply(0,5)=0 |
testIsPalindrome | "racecar"=true, "hello"=false, "A man a plan..."=true |
testFibonacci | fib(0)=0, fib(1)=1, fib(10)=55 |
Python 测试 (36 个, Pytest)
| 测试类 | 测试数 | 覆盖内容 |
|---|---|---|
TestCalculator | 10 | add/subtract/multiply/divide/power/modulo/history/clear_history |
TestIsPalindrome | 5 | 简单回文/非回文/大小写/特殊字符/空字符串 |
TestWordFrequency | 3 | 基础/大小写/空文本 |
TestFlatten | 4 | 简单/嵌套/空/已扁平 |
TestCamelToSnake | 3 | 基础/单词/缩写词 |
TestSnakeToCamel | 2 | 基础/单词 |
TestSafeDivide | 3 | 正常/零除/类型错误 |
TestChunkList | 3 | 基础/精确分块/非法大小 |
TestDeepGet | 3 | 基础/缺失键/顶层 |
文档生成时间: 2026-08-25 | 最终成功构建: Build #12 (103s)