AI系统陪跑：构建软件全生命周期智能辅助引擎

import chromadb
from sentence_transformers import SentenceTransformer

class KnowledgeBase:
    def __init__(self, path="./kb_store"):
        self.client = chromadb.PersistentClient(path)
        self.collection = self.client.get_or_create_collection("dev_knowledge")
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
    
    def add_doc(self, doc_id: str, content: str, metadata: dict):
        emb = self.encoder.encode(content).tolist()
        self.collection.add(documents=[content], embeddings=[emb], ids=[doc_id], metadatas=[metadata])
    
    def retrieve(self, query: str, top_k=3):
        q_emb = self.encoder.encode(query).tolist()
        results = self.collection.query(query_embeddings=[q_emb], n_results=top_k)
        return [{"content": c, "meta": m} for c, m in zip(results['documents'][0], results['metadatas'][0])]

# 初始化并添加示例规范
kb = KnowledgeBase()
kb.add_doc("style_001", "变量命名采用小驼峰，禁止使用拼音", {"type": "style"})
kb.add_doc("bug_001", "空指针异常通常发生在未判空的对象调用，建议使用Optional或提前检查", {"type": "bug"})

from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
import json

class CodeReviewer:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
        self.kb = KnowledgeBase()
    
    def review(self, diff_text: str, repo_context: str = "") -> list:
        # 提取关键词以检索知识库
        query = f"代码变更涉及：{diff_text[:200]}"
        docs = self.kb.retrieve(query, top_k=2)
        kb_context = "\n".join([d['content'] for d in docs])
        
        prompt = PromptTemplate.from_template("""
你是资深代码审查专家，请对以下代码变更进行审查。
仓库背景：{repo_context}
知识库参考：{kb_context}
变更内容：
{diff_text}
输出JSON数组，每项包含：file, line_range, severity, suggestion, reference。
""")
        chain = prompt | self.llm
        resp = chain.invoke({"repo_context": repo_context, "kb_context": kb_context, "diff_text": diff_text})
        try:
            return json.loads(resp.content)
        except:
            return [{"error": "解析失败", "raw": resp.content}]

# 示例：模拟git diff
diff = """
- public void process(String name) {
-    if (name.equals("admin")) { ... }
- }
+ public void process(String name) {
+    if ("admin".equals(name)) { ... }
+ }
"""
reviewer = CodeReviewer()
comments = reviewer.review(diff)
print(comments)

import re
from datetime import datetime

class LogAnalyzer:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o-mini")
        self.kb = KnowledgeBase()
    
    def analyze(self, log_lines: list[str]) -> dict:
        # 提取错误堆栈
        errors = [line for line in log_lines if re.search(r'ERROR|Exception|FATAL', line)]
        if not errors:
            return {"status": "ok"}
        
        combined = "\n".join(errors[-20:])  # 取最近20条
        query = f"错误日志：{combined[:300]}"
        docs = self.kb.retrieve(query)
        kb_info = "\n".join([d['content'] for d in docs])
        
        prompt = f"""
你是一个运维专家，根据以下错误日志和历史知识，分析可能原因和修复建议。
日志：
{combined}
历史知识：
{kb_info}
输出JSON：{{"root_cause": "...", "suggestion": "...", "severity": "high/medium/low"}}
"""
        resp = self.llm.invoke(prompt)
        return json.loads(resp.content)

import requests

def create_issue(title, description, severity):
    # 示例：发送至飞书
    webhook = "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"
    msg = {
        "msg_type": "interactive",
        "card": {
            "header": {"title": {"tag": "plain_text", "content": f"[{severity}] {title}"}},
            "body": [{"tag": "div", "text": {"content": description}}]
        }
    }
    requests.post(webhook, json=msg)

stage('AI陪跑审查') {
    steps {
        script {
            def diff = sh(script: 'git diff HEAD~1', returnStdout: true)
            def result = python3("""
from code_reviewer import CodeReviewer
cr = CodeReviewer()
comments = cr.review('''$diff''')
print(comments)
            """)
            // 若存在严重问题，构建失败并输出报告
        }
    }
}