필사 모드: AI 创业公司与产品开发指南：从 LLM API 到扩展、商业模式

# 安装 Ollama 后运行模型
ollama pull llama3.3
ollama run llama3.3

pip install vllm

python -m vllm.entrypoints.openai.api_server \
  --model meta-llama/Llama-3.3-70B-Instruct \
  --tensor-parallel-size 4 \
  --gpu-memory-utilization 0.9

预算是否有限?
├── 是 → 开源(Llama、Mistral)自托管
│         或 Gemini Flash(低成本 API)
└── 否 → 质量要求是否高?
          ├── 是 → Claude 3.5 Sonnet / GPT-4o
          └── 否 → GPT-4o mini / Claude Haiku

import streamlit as st
from openai import OpenAI

client = OpenAI()

st.title("AI 文档摘要器 MVP")

uploaded_file = st.file_uploader("上传文档", type=["txt", "pdf"])
tone = st.selectbox("摘要语气", ["商务", "轻松", "技术"])

if uploaded_file and st.button("开始摘要"):
    text = uploaded_file.read().decode("utf-8")

    with st.spinner("AI 正在生成摘要..."):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": f"你是一名专业的文档摘要撰写者。请使用{tone}的语气进行摘要。"
                },
                {
                    "role": "user",
                    "content": f"请将以下文档摘要为 3-5 句话:\n\n{text[:4000]}"
                }
            ],
            max_tokens=500
        )

    summary = response.choices[0].message.content
    st.success("摘要完成!")
    st.write(summary)

    st.download_button(
        label="下载摘要",
        data=summary,
        file_name="summary.txt"
    )

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA

# 文档分块
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(documents)

# 创建向量存储
vectorstore = Chroma.from_documents(
    chunks,
    OpenAIEmbeddings()
)

# 构建 RAG 链
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

result = qa_chain.invoke({"query": "产品的退款政策是什么？"})
print(result["result"])

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

judge_llm = ChatOpenAI(model="gpt-4o", temperature=0)

EVAL_PROMPT = ChatPromptTemplate.from_template("""
你是一名 AI 响应质量评估者。

问题: {question}
AI 响应: {response}
参考答案: {reference}

请按以下标准评分(1-5 分):
- 准确性 (Accuracy): 是否事实准确?
- 完整性 (Completeness): 是否充分回答了问题?
- 清晰度 (Clarity): 是否易于理解?

以 JSON 格式回复: {{"accuracy": 分数, "completeness": 分数, "clarity": 分数, "reasoning": "说明"}}
""")

def evaluate_response(question, response, reference):
    chain = EVAL_PROMPT | judge_llm
    result = chain.invoke({
        "question": question,
        "response": response,
        "reference": reference
    })
    return result.content

import random
from dataclasses import dataclass
from typing import Dict, List
import json

@dataclass
class PromptVariant:
    name: str
    system_prompt: str
    wins: int = 0
    total: int = 0

    @property
    def win_rate(self):
        return self.wins / self.total if self.total > 0 else 0

class PromptABTest:
    def __init__(self, variants: List[PromptVariant]):
        self.variants = {v.name: v for v in variants}

    def select_variant(self) -> PromptVariant:
        # ε-贪心：10% 探索，90% 利用
        if random.random() < 0.1:
            return random.choice(list(self.variants.values()))
        return max(self.variants.values(), key=lambda v: v.win_rate)

    def record_feedback(self, variant_name: str, positive: bool):
        v = self.variants[variant_name]
        v.total += 1
        if positive:
            v.wins += 1

    def report(self) -> Dict:
        return {
            name: {"win_rate": f"{v.win_rate:.1%}", "total": v.total}
            for name, v in self.variants.items()
        }

# 使用示例
ab_test = PromptABTest([
    PromptVariant("formal", "你是一名专业且正式的 AI 助手。"),
    PromptVariant("casual", "你好！我是一个用轻松易懂的方式讲解的 AI。"),
])

from fastapi import FastAPI
from pydantic import BaseModel
from datetime import datetime
import json

app = FastAPI()

class FeedbackRequest(BaseModel):
    session_id: str
    message_id: str
    rating: int  # 1-5
    comment: str = ""
    prompt_variant: str = "default"

feedback_store = []

@app.post("/feedback")
async def collect_feedback(feedback: FeedbackRequest):
    entry = {
        "timestamp": datetime.utcnow().isoformat(),
        **feedback.dict()
    }
    feedback_store.append(entry)

    # 实际环境中应存入数据库
    with open("feedback_log.jsonl", "a") as f:
        f.write(json.dumps(entry) + "\n")

    return {"status": "recorded", "message_id": feedback.message_id}

@app.get("/feedback/stats")
async def get_stats():
    if not feedback_store:
        return {"avg_rating": 0, "total": 0}

    avg = sum(f["rating"] for f in feedback_store) / len(feedback_store)
    return {
        "avg_rating": round(avg, 2),
        "total": len(feedback_store),
        "positive_rate": f"{sum(1 for f in feedback_store if f['rating'] >= 4) / len(feedback_store):.1%}"
    }

from dataclasses import dataclass
from typing import Dict

@dataclass
class ModelPricing:
    input_per_1m: float   # USD per 1M input tokens
    output_per_1m: float  # USD per 1M output tokens
    cache_write_per_1m: float = 0.0
    cache_read_per_1m: float = 0.0

PRICING: Dict[str, ModelPricing] = {
    "gpt-4o": ModelPricing(2.50, 10.00),
    "gpt-4o-mini": ModelPricing(0.15, 0.60),
    "claude-3-5-sonnet": ModelPricing(3.00, 15.00, 3.75, 0.30),
    "claude-3-haiku": ModelPricing(0.25, 1.25, 0.30, 0.03),
    "gemini-2.0-flash": ModelPricing(0.075, 0.30),
}

def calculate_monthly_cost(
    model: str,
    daily_requests: int,
    avg_input_tokens: int,
    avg_output_tokens: int,
    cache_hit_rate: float = 0.0
) -> dict:
    p = PRICING[model]
    monthly_requests = daily_requests * 30

    # 考虑缓存命中后的实际输入 token
    cached_tokens = avg_input_tokens * cache_hit_rate
    fresh_tokens = avg_input_tokens * (1 - cache_hit_rate)

    input_cost = (fresh_tokens * monthly_requests / 1_000_000) * p.input_per_1m
    cache_read_cost = (cached_tokens * monthly_requests / 1_000_000) * p.cache_read_per_1m
    output_cost = (avg_output_tokens * monthly_requests / 1_000_000) * p.output_per_1m

    total = input_cost + cache_read_cost + output_cost

    return {
        "model": model,
        "monthly_requests": monthly_requests,
        "total_usd": round(total, 2),
        "cost_per_request_usd": round(total / monthly_requests, 6),
        "breakdown": {
            "input": round(input_cost, 2),
            "cache_read": round(cache_read_cost, 2),
            "output": round(output_cost, 2)
        }
    }

# 示例：每天 10,000 次请求
result = calculate_monthly_cost(
    model="claude-3-5-sonnet",
    daily_requests=10_000,
    avg_input_tokens=2000,
    avg_output_tokens=500,
    cache_hit_rate=0.7  # 70% 缓存命中率
)
print(f"月预估成本: ${result['total_usd']}")

import anthropic

client = anthropic.Anthropic()

# 将系统提示词存入缓存
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是一名专业的法律 AI 助手。",
        },
        {
            "type": "text",
            "text": open("legal_knowledge_base.txt").read(),  # 大容量上下文
            "cache_control": {"type": "ephemeral"}  # 标记为缓存
        }
    ],
    messages=[
        {"role": "user", "content": "请说明合同解除的条件。"}
    ]
)

# 通过 cache_read_input_tokens 确认缓存效率
usage = response.usage
print(f"缓存读取 token: {usage.cache_read_input_tokens}")
print(f"缓存写入 token: {usage.cache_creation_input_tokens}")

from openai import OpenAI

client = OpenAI()

def grounded_response(query: str, context: str) -> dict:
    """基于 RAG 生成有依据的响应"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """请务必仅根据提供的上下文作答。
                若上下文中没有相关信息，请明确说明"提供的信息中没有该内容"。
                若不确定，请回答"需要进一步确认"。"""
            },
            {
                "role": "user",
                "content": f"上下文:\n{context}\n\n问题: {query}"
            }
        ],
        temperature=0,  # 确定性响应
        logprobs=True   # 用于确认置信度
    )

    message = response.choices[0]
    # 用 logprob 估计不确定性
    avg_logprob = sum(
        t.logprob for t in message.logprobs.content
    ) / len(message.logprobs.content) if message.logprobs else 0

    return {
        "answer": message.message.content,
        "confidence": round(2.718 ** avg_logprob, 3),  # e^logprob = prob
        "needs_review": avg_logprob < -1.5  # 低置信度标记
    }