AI 时代生存指南第 2 篇：从开发者转型为 LLM 工程师/AI 工程师

# OpenAI API 基本用法
from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一名有帮助的代码评审员。"},
        {"role": "user", "content": "请评审这个 Python 函数：def add(a, b): return a + b"}
    ],
    temperature=0.7,
    max_tokens=500
)

print(response.choices[0].message.content)
print(f"使用的 token 数: {response.usage.total_tokens}")
print(f"成本 (按 gpt-4o 计费): ${response.usage.total_tokens * 0.000005:.6f}")

# 结构化输出示例
from openai import OpenAI
from pydantic import BaseModel
from typing import List

class CodeReview(BaseModel):
    issues: List[str]
    suggestions: List[str]
    severity: str  # "low", "medium", "high"
    score: int     # 1-10

client = OpenAI()

response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一名资深开发者。请评审代码并以 JSON 格式返回结果。"},
        {"role": "user", "content": "请评审以下代码：[代码内容]"}
    ],
    response_format=CodeReview
)

review = response.choices[0].message.parsed
print(f"严重程度: {review.severity}")
print(f"评分: {review.score}/10")

# 用 LangChain 构建 RAG 系统
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 1. 加载文档
loader = PyPDFLoader("company_docs.pdf")
documents = loader.load()

# 2. 分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)

# 3. 嵌入并存入向量数据库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. 创建 RAG 链
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

# 5. 提问
result = qa_chain.invoke({"query": "我们公司的年假政策是什么?"})
print(result["result"])

# LangChain Agent 示例：代码评审智能体
from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_community.utilities import SerpAPIWrapper

# 定义工具
def analyze_code_complexity(code: str) -> str:
    """分析代码的复杂度。"""
    # 实际实现中会使用 radon 等库
    lines = code.count('\n')
    functions = code.count('def ')
    return f"代码行数: {lines}, 函数数量: {functions}, 复杂度得分: {lines // 10}"

def check_security_issues(code: str) -> str:
    """检查基本的安全漏洞。"""
    issues = []
    if 'eval(' in code:
        issues.append("检测到 eval() 用法 - 安全风险")
    if 'shell=True' in code:
        issues.append("检测到 shell=True 用法 - 命令注入风险")
    if not issues:
        return "未发现明显的安全漏洞"
    return "\n".join(issues)

tools = [
    Tool(
        name="代码复杂度分析",
        func=analyze_code_complexity,
        description="分析 Python 代码的复杂度。输入: 代码字符串"
    ),
    Tool(
        name="安全漏洞检查",
        func=check_security_issues,
        description="检查代码中的安全漏洞。输入: 代码字符串"
    )
]

llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "请对以下代码进行综合评审: [代码内容]"
})

from crewai import Agent, Task, Crew

# 定义智能体
researcher = Agent(
    role="技术调研员",
    goal="调研给定主题的最新技术趋势",
    backstory="拥有 10 年经验的技术分析师",
    verbose=True
)

writer = Agent(
    role="技术博客作者",
    goal="把调研结果写成通俗易懂的博客文章",
    backstory="专注于开发者内容的技术博客专业作者",
    verbose=True
)

# 定义任务
research_task = Task(
    description="调研 LLM 工程领域的最新趋势",
    agent=researcher
)

writing_task = Task(
    description="基于调研结果撰写博客文章",
    agent=writer
)

# 执行 crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task]
)
result = crew.kickoff()

from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer
from datasets import load_dataset

# 加载基础模型
model_name = "beomi/Llama-3-Open-Ko-8B"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    load_in_4bit=True,  # QLoRA: 4 位量化
    device_map="auto"
)

# LoRA 配置
lora_config = LoraConfig(
    r=16,           # 秩(rank)，越低参数量越少
    lora_alpha=32,  # 缩放参数
    target_modules=["q_proj", "v_proj"],  # 仅应用于 attention 层
    lora_dropout=0.05,
    bias="none",
    task_type=TaskType.CAUSAL_LM
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出示例: trainable params: 3,407,872 (占全体的 0.04%)

from ragas import evaluate
from ragas.metrics import (
    faithfulness,        # 答案是否忠实于上下文
    answer_relevancy,    # 答案是否与问题相关
    context_precision,   # 检索到的上下文是否准确
    context_recall       # 是否检索到了全部所需的上下文
)

# 评估数据集
from datasets import Dataset

data = {
    "question": ["公司的年假政策是什么?", "工资什么时候发放?"],
    "answer": ["提供 15 天年假。", "每月 25 日发放。"],
    "contexts": [
        ["每年提供 15 天年假，剩余年假不可结转..."],
        ["工资发放日为每月 25 日，如遇节假日则提前一天发放..."]
    ],
    "ground_truth": ["每年 15 天", "每月 25 日"]
}

dataset = Dataset.from_dict(data)
result = evaluate(dataset, metrics=[faithfulness, answer_relevancy])
print(result)

# 安装 vLLM 并部署模型
pip install vllm

# 启动 API 服务器
python -m vllm.entrypoints.openai.api_server \
  --model meta-llama/Llama-3-8B-Instruct \
  --max-model-len 4096 \
  --gpu-memory-utilization 0.9

# 成本监控包装器示例
import time
from functools import wraps
from dataclasses import dataclass

COST_PER_TOKEN = {
    "gpt-4o": {"input": 0.000005, "output": 0.000015},
    "gpt-4o-mini": {"input": 0.00000015, "output": 0.0000006},
    "claude-3-5-sonnet": {"input": 0.000003, "output": 0.000015},
}

@dataclass
class APICall:
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    cost_usd: float

def track_llm_cost(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        elapsed = (time.time() - start) * 1000

        # 使用量追踪 (实际实现中会存入数据库)
        print(f"延迟: {elapsed:.0f}ms")
        return result
    return wrapper

import os
from langchain_core.tracers import LangChainTracer

# 设置环境变量
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"

# 此后所有 LangChain 调用都会被自动追踪

pip install openai langchain chromadb tiktoken

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import TextLoader
import os

os.environ["OPENAI_API_KEY"] = "your-api-key"

# 加载图书信息
loader = TextLoader("books.txt", encoding="utf-8")
documents = loader.load()

# 分块
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)

# 创建向量数据库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="books_db"
)
print(f"已存储的分块数: {len(chunks)}")

from langchain_openai import ChatOpenAI
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

# LLM 配置
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 对话记忆
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# RAG 链
qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    memory=memory,
    verbose=True
)

# 对话循环
print("这是图书推荐聊天机器人。您在寻找什么类型的书?")
while True:
    query = input("You: ")
    if query.lower() in ["quit", "exit", "退出"]:
        break

    response = qa_chain.invoke({"question": query})
    print(f"Bot: {response['answer']}\n")