AGI大模型应用：从模型到产品的工程化蜕变

import os
from typing import List
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# 1. 加载文档
loader = PyPDFLoader("./knowledge_base/运维手册.pdf")
documents = loader.load()

# 2. 语义切分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
)
chunks = text_splitter.split_documents(documents)

# 3. 向量化存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")

# 4. 检索器配置：混合检索（需要额外安装 BM25 插件）
retriever = vectorstore.as_retriever(
    search_type="mmr",          # 最大边际相关性，提升多样性
    search_kwargs={"k": 5, "fetch_k": 20}
)

# 5. 构建 QA 链
llm = ChatOpenAI(model_name="gpt-4", temperature=0.2)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 执行查询
result = qa_chain("CPU 使用率持续超过 90% 应该如何处理？")
print(f"答案: {result['result']}")
print(f"参考文档: {[doc.metadata for doc in result['source_documents']]}")

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgentState(TypedDict):
    question: str
    retrieved_docs: List[str]
    rounds: int
    answer: str

def retrieve(state: AgentState):
    # 初次检索
    docs = vectorstore.similarity_search(state["question"], k=5)
    return {"retrieved_docs": [d.page_content for d in docs], "rounds": 1}

def evaluate(state: AgentState):
    # 用大模型评估检索结果是否充分
    prompt = f"问题: {state['question']}\n已检索内容: {state['retrieved_docs']}\n这些信息足够回答吗？回复'是'或'否'及理由。"
    response = llm.invoke(prompt)
    if "是" in response.content and state["rounds"] >= 1:
        return "generate"
    elif state["rounds"] >= 3:
        return "generate"  # 最多 3 轮
    else:
        return "refine"

def refine_query(state: AgentState):
    # 改写查询词，补充检索
    prompt = f"原始问题: {state['question']}\n已有内容: {state['retrieved_docs']}\n请生成一个更精准的检索查询。"
    new_query = llm.invoke(prompt).content
    docs = vectorstore.similarity_search(new_query, k=3)
    state["retrieved_docs"].extend([d.page_content for d in docs])
    state["rounds"] += 1
    return state

def generate(state: AgentState):
    prompt = f"基于以下信息回答问题:\n{chr(10).join(state['retrieved_docs'])}\n\n问题: {state['question']}"
    answer = llm.invoke(prompt).content
    return {"answer": answer}

# 构建状态图
graph = StateGraph(AgentState)
graph.add_node("retrieve", retrieve)
graph.add_node("evaluate", evaluate)
graph.add_node("refine", refine_query)
graph.add_node("generate", generate)

graph.set_entry_point("retrieve")
graph.add_conditional_edges("evaluate", lambda s: s["next_step"])
graph.add_edge("refine", "evaluate")
graph.add_edge("generate", END)

app = graph.compile()
result = app.invoke({"question": "如何排查数据库连接池耗尽问题？", "rounds": 0})
print(result["answer"])

import org.springframework.ai.chat.ChatClient;
import org.springframework.ai.chat.ChatResponse;
import org.springframework.ai.chat.prompt.Prompt;
import org.springframework.ai.chat.prompt.SystemPromptTemplate;
import org.springframework.ai.tool.annotation.Tool;
import org.springframework.ai.tool.annotation.ToolParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

@Service
public class AgentService {

    @Autowired
    private ChatClient chatClient;

    @Tool(description = "查询指定时间范围内的订单数量")
    public int queryOrderCount(
            @ToolParam(description = "开始时间，格式 yyyy-MM-dd HH:mm:ss") String startTime,
            @ToolParam(description = "结束时间，格式 yyyy-MM-dd HH:mm:ss") String endTime) {
        // 实际场景中这里会查询数据库
        System.out.println("查询订单: " + startTime + " 到 " + endTime);
        return 1342;  // 模拟返回
    }

    @Tool(description = "获取当前系统时间")
    public String getCurrentTime() {
        return LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
    }

    public String processUserRequest(String userInput) {
        // 构建系统提示词，声明可用工具
        String systemPrompt = """
                你是一个智能运维助手，你可以使用以下工具：
                1. queryOrderCount - 查询订单数量
                2. getCurrentTime - 获取当前时间
                请根据用户问题，自主决定调用哪些工具，并基于结果给出最终答案。
                """;

        Prompt prompt = new Prompt(
                new SystemPromptTemplate(systemPrompt).createMessage(),
                new UserMessage(userInput)
        );

        // Spring AI 会自动将 @Tool 注解的方法注册给模型
        ChatResponse response = chatClient.call(prompt);
        return response.getResult().getOutput().getContent();
    }
}

def react_loop(user_question, max_steps=5):
    thought_history = []
    for step in range(max_steps):
        # 1. 推理：决定下一步动作
        prompt = f"""
        你是一个 Agent，当前任务：{user_question}
        历史操作：{thought_history}
        可选工具：search(query), calculate(expr), get_weather(city)
        请输出下一步行动，格式：Action: tool_name(parameters)
        如果已经有答案，输出：Final Answer: 答案内容
        """
        response = llm.invoke(prompt).content
        
        if "Final Answer:" in response:
            return response.split("Final Answer:")[1].strip()
        
        # 2. 解析动作并执行
        action = parse_action(response)  # 解析出 tool_name 和参数
        result = execute_tool(action.tool, action.params)
        
        # 3. 观察结果，加入历史
        thought_history.append(f"执行 {action.tool} 得到结果: {result}")
    
    return "抱歉，未能完成任务。"

# 使用示例
print(react_loop("帮我查一下北京今天的天气，然后计算比昨天高多少度？"))

# 使用 RAGAS 对 RAG 系统进行自动评估
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall
from datasets import Dataset

# 准备测试数据
test_data = Dataset.from_dict({
    "question": [
        "什么是 Kubernetes 的 Pod？",
        "如何排查内存泄漏？"
    ],
    "answer": [
        qa_chain("什么是 Kubernetes 的 Pod？")["result"],
        qa_chain("如何排查内存泄漏？")["result"]
    ],
    "contexts": [
        ["Kubernetes Pod 是...", "Pod 是 K8s 最小部署单元..."],
        ["内存泄漏通常表现为...", "使用 jstat 监控 GC 频率..."]
    ]
})

# 运行评估
result = evaluate(
    dataset=test_data,
    metrics=[faithfulness, answer_relevancy, context_recall]
)
print(result)

# 将结果接入 CI：如果 faithfulness < 0.8，构建失败
if result["faithfulness"] < 0.8:
    raise Exception("RAG 质量不达标，请检查检索效果")

# .github/workflows/rag_eval.yml
name: RAG Quality Check
on: [push, pull_request]

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - run: pip install -r requirements.txt
      - run: python tests/evaluate_rag.py
      - name: 检查评估结果
        run: |
          if [ $(cat eval_result.json | jq '.faithfulness') -lt 0.8 ]; then
            echo "Faithfulness 低于阈值 0.8，请优化 RAG 流水线"
            exit 1
          fi

import hashlib
from datetime import datetime

class CostTracker:
    def __init__(self):
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.input_price_per_1k = 0.01   # 以 GPT-4 为例
        self.output_price_per_1k = 0.03
        
    def log_call(self, prompt: str, response: str, model: str = "gpt-4"):
        # 粗略估计 Token 数：中文约 1.5 字符/Token，英文约 4 字符/Token
        input_tokens = len(prompt) // 2
        output_tokens = len(response) // 2
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
        
        cost = (input_tokens / 1000 * self.input_price_per_1k + 
                output_tokens / 1000 * self.output_price_per_1k)
        print(f"[{datetime.now()}] 本次调用成本: ${cost:.4f}")
        return cost
    
    def report(self):
        total = (self.total_input_tokens / 1000 * self.input_price_per_1k + 
                 self.total_output_tokens / 1000 * self.output_price_per_1k)
        print(f"总输入 Token: {self.total_input_tokens}")
        print(f"总输出 Token: {self.total_output_tokens}")
        print(f"总成本: ${total:.2f}")

# 使用
tracker = CostTracker()
response = qa_chain("什么是智能运维？")
tracker.log_call("什么是智能运维？", response["result"])

def route_model(user_question: str) -> str:
    """根据问题复杂度选择模型"""
    # 用轻量级分类器判断问题复杂度
    simple_keywords = ["是什么", "定义", "多少", "几", "哪个"]
    complex_keywords = ["为什么", "如何", "比较", "分析", "优化", "设计"]
    
    if any(k in user_question for k in simple_keywords) and \
       not any(k in user_question for k in complex_keywords):
        return "gpt-3.5-turbo"   # 低成本模型
    else:
        return "gpt-4"          # 高性能模型

# 调用时动态选择
model = route_model(user_input)
response = chat_completion(model=model, messages=[...])

// 使用 Spring Retry 实现模型调用降级
@Retryable(
    value = {TimeoutException.class},
    maxAttempts = 2,
    backoff = @Backoff(delay = 1000)
)
public String callModelWithFallback(String prompt) {
    try {
        return primaryModel.generate(prompt);
    } catch (Exception e) {
        // 降级到备用模型
        return fallbackModel.generate(prompt);
    }
}

// 备用模型返回兜底响应
@Recover
public String recover(TimeoutException e, String prompt) {
    return "抱歉，当前模型服务繁忙，您的请求已转交人工处理，工单号: " + UUID.randomUUID();
}

import structlog

logger = structlog.get_logger()

def call_with_observability(prompt, model="gpt-4"):
    start = time.time()
    try:
        response = llm.invoke(prompt)
        latency = time.time() - start
        logger.info(
            "model_call",
            model=model,
            prompt_length=len(prompt),
            response_length=len(response.content),
            latency_ms=latency * 1000,
            status="success"
        )
        return response
    except Exception as e:
        logger.error("model_call_failed", model=model, error=str(e))
        raise

import json
from uuid import uuid4

class Trace:
    def __init__(self, session_id):
        self.session_id = session_id
        self.steps = []
    
    def add_step(self, step_name, input_data, output_data, metadata=None):
        self.steps.append({
            "step": step_name,
            "input": input_data[:200],   # 截断防止过大
            "output": output_data[:200],
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat()
        })
    
    def save(self):
        with open(f"traces/{self.session_id}.json", "w") as f:
            json.dump(self.steps, f, indent=2, ensure_ascii=False)

# 在 Agent 中使用
trace = Trace(session_id=uuid4().hex)
trace.add_step("retrieve", question, retrieved_docs, {"k": 5})
trace.add_step("generate", prompt, answer, {"model": "gpt-4"})
trace.save()