导读：本文通过一个完整的实战项目——智能文档问答系统，带你掌握GPT/Claude API的核心用法。从环境搭建到功能实现，再到成本优化，手把手教你开发一个可落地的AI应用。

---

📑 文章目录

1. 项目概述与技术选型
2. 环境搭建：5分钟快速开始
3. 核心功能实现
4. 完整代码实现
5. 成本优化技巧
6. 部署与测试
7. 常见问题解决

---

一、项目概述与技术选型 🎯

1.1 项目介绍

智能文档问答系统：上传文档（PDF/TXT/Markdown），AI自动理解内容并回答相关问题。

应用场景：

• 企业知识库问答
• 技术文档查询
• 合同/报告分析
• 学习资料辅导

1.2 技术架构

┌─────────────────────────────────────┐
│         前端交互层                   │
│    (Streamlit / Flask)              │
└─────────────────────────────────────┘
              ↓
┌─────────────────────────────────────┐
│        文档处理层                    │
│   PyPDF2 / python-docx              │
│   文本分块 & 向量化                  │
└─────────────────────────────────────┘
              ↓
┌─────────────────────────────────────┐
│        向量存储层                    │
│   FAISS / ChromaDB                  │
│   语义搜索 & 相关内容检索            │
└─────────────────────────────────────┘
              ↓
┌─────────────────────────────────────┐
│       AI推理层                       │
│   OpenAI API / Claude API           │
│   基于检索内容生成答案               │
└─────────────────────────────────────┘

1.3 技术选型

API选择：OpenAI GPT-4

• 代码能力强
• 生态成熟
• 文档完善

核心库：

openai==1.12.0          # OpenAI API
langchain==0.1.0        # AI应用框架
faiss-cpu==1.7.4        # 向量检索
pypdf2==3.0.1           # PDF处理
streamlit==1.31.0       # Web界面
python-dotenv==1.0.0    # 环境变量

---

二、环境搭建：5分钟快速开始 🚀

2.1 安装依赖

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install openai langchain faiss-cpu pypdf2 streamlit python-dotenv tiktoken

2.2 配置API Key

# 创建 .env 文件
echo "OPENAI_API_KEY=sk-your-api-key-here" > .env

2.3 项目结构

document-qa/
├── .env                    # API密钥
├── app.py                  # 主程序
├── utils/
│   ├── document_loader.py  # 文档加载
│   ├── text_splitter.py    # 文本分块
│   └── qa_chain.py         # 问答链
├── data/                   # 上传的文档
└── requirements.txt        # 依赖列表

---

三、核心功能实现 💻

3.1 文档加载与处理

document_loader.py

from PyPDF2 import PdfReader
from typing import List

def load_pdf(file_path: str) -> str:
    """加载PDF文件并提取文本"""
    reader = PdfReader(file_path)
    text = ""
    for page in reader.pages:
        text += page.extract_text()
    return text

def load_txt(file_path: str) -> str:
    """加载TXT文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        return f.read()

3.2 文本分块策略

text_splitter.py

from typing import List

def split_text(text: str, chunk_size: int = 1000, overlap: int = 200) -> List[str]:
    """
    将长文本分割成小块，保持语义连贯性
    
    Args:
        text: 原始文本
        chunk_size: 每块大小（字符数）
        overlap: 重叠部分大小
    """
    chunks = []
    start = 0
    
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start += chunk_size - overlap
    
    return chunks

# 示例
text = "这是一篇很长的文档..." * 100
chunks = split_text(text, chunk_size=500, overlap=50)
print(f"分割成 {len(chunks)} 个块")

为什么要分块？

问题：GPT-4有128K token限制，长文档无法一次处理

解决方案：
1. 将文档分成小块
2. 用户提问时，只检索相关块
3. 只发送相关内容给API

优势：
✓ 突破token限制
✓ 降低API成本
✓ 提高响应速度
✓ 答案更精准

3.3 向量检索系统

使用FAISS构建向量数据库

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
import os

class DocumentVectorStore:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            openai_api_key=os.getenv("OPENAI_API_KEY")
        )
        self.vector_store = None
    
    def create_from_text(self, text: str):
        """从文本创建向量库"""
        # 智能分块
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", "。", "！", "？", "；", " "]
        )
        chunks = splitter.split_text(text)
        
        # 创建向量库
        self.vector_store = FAISS.from_texts(
            texts=chunks,
            embedding=self.embeddings
        )
        
        print(f"✓ 向量库创建成功，包含 {len(chunks)} 个文本块")
    
    def search(self, query: str, k: int = 3) -> List[str]:
        """语义搜索相关内容"""
        if not self.vector_store:
            return []
        
        docs = self.vector_store.similarity_search(query, k=k)
        return [doc.page_content for doc in docs]

工作原理图解

用户问题："什么是Transformer？"
        ↓
    向量化 (Embedding)
        ↓
    [0.23, -0.45, 0.78, ...]  (1536维向量)
        ↓
    在向量库中搜索相似向量
        ↓
┌─────────────────────────────────┐
│ 文档块1："Transformer是..."      │  相似度: 0.92 ⭐
│ 文档块2："注意力机制..."        │  相似度: 0.85 ⭐
│ 文档块3："神经网络..."          │  相似度: 0.45
└─────────────────────────────────┘
        ↓
    返回Top 3最相关块

3.4 问答链实现

qa_chain.py

from openai import OpenAI
import os

class QAChain:
    def __init__(self):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    def answer(self, question: str, context: List[str]) -> str:
        """
        基于检索到的上下文回答问题
        
        Args:
            question: 用户问题
            context: 检索到的相关文档片段
        """
        # 组合上下文
        combined_context = "\n\n---\n\n".join(context)
        
        # 构建提示词
        prompt = f"""基于以下文档内容回答问题。如果文档中没有相关信息，请明确说明。

文档内容：
{combined_context}

问题：{question}

请给出准确、简洁的回答："""
        
        # 调用API
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是一个专业的文档分析助手，根据提供的文档内容准确回答问题。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,  # 降低随机性，提高准确性
            max_tokens=500
        )
        
        return response.choices[0].message.content

---

四、完整代码实现 🔧

4.1 主程序 app.py

import streamlit as st
from utils.document_loader import load_pdf, load_txt
from utils.qa_chain import QAChain, DocumentVectorStore
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 页面配置
st.set_page_config(
    page_title="智能文档问答系统",
    page_icon="📚",
    layout="wide"
)

# 标题
st.title("📚 智能文档问答系统")
st.markdown("上传文档，AI帮你快速找答案！")

# 初始化session state
if 'vector_store' not in st.session_state:
    st.session_state.vector_store = None
if 'qa_chain' not in st.session_state:
    st.session_state.qa_chain = QAChain()
if 'chat_history' not in st.session_state:
    st.session_state.chat_history = []

# 侧边栏：文档上传
with st.sidebar:
    st.header("📄 文档上传")
    uploaded_file = st.file_uploader(
        "选择文件",
        type=['pdf', 'txt'],
        help="支持PDF和TXT格式"
    )
    
    if uploaded_file:
        with st.spinner("正在处理文档..."):
            # 保存文件
            file_path = f"data/{uploaded_file.name}"
            os.makedirs("data", exist_ok=True)
            with open(file_path, "wb") as f:
                f.write(uploaded_file.getbuffer())
            
            # 加载文档
            if uploaded_file.name.endswith('.pdf'):
                text = load_pdf(file_path)
            else:
                text = load_txt(file_path)
            
            # 创建向量库
            vector_store = DocumentVectorStore()
            vector_store.create_from_text(text)
            st.session_state.vector_store = vector_store
            
            st.success(f"✓ 文档已加载！共 {len(text)} 字符")
    
    # 显示统计
    if st.session_state.vector_store:
        st.info(f"💬 已回答 {len(st.session_state.chat_history)} 个问题")

# 主区域：问答界面
if st.session_state.vector_store is None:
    st.info("👈 请先上传文档")
else:
    # 显示对话历史
    for qa in st.session_state.chat_history:
        with st.chat_message("user"):
            st.write(qa['question'])
        with st.chat_message("assistant"):
            st.write(qa['answer'])
    
    # 问题输入
    question = st.chat_input("输入你的问题...")
    
    if question:
        # 显示用户问题
        with st.chat_message("user"):
            st.write(question)
        
        # 生成回答
        with st.chat_message("assistant"):
            with st.spinner("思考中..."):
                # 检索相关内容
                relevant_docs = st.session_state.vector_store.search(question, k=3)
                
                # 生成答案
                answer = st.session_state.qa_chain.answer(question, relevant_docs)
                
                st.write(answer)
                
                # 显示引用来源
                with st.expander("📖 查看相关文档片段"):
                    for i, doc in enumerate(relevant_docs, 1):
                        st.markdown(f"**片段 {i}:**")
                        st.text(doc[:200] + "...")
        
        # 保存对话历史
        st.session_state.chat_history.append({
            'question': question,
            'answer': answer
        })

4.2 运行项目

# 启动应用
streamlit run app.py

# 浏览器自动打开 http://localhost:8501

4.3 使用流程

步骤1：上传文档
→ 点击侧边栏"Browse files"
→ 选择PDF或TXT文件
→ 等待处理完成

步骤2：提问
→ 在底部输入框输入问题
→ 例如："文档的主要观点是什么？"

步骤3：查看答案
→ AI自动回答
→ 点击"查看相关文档片段"看引用来源

---

五、成本优化技巧 💰

5.1 Token使用优化

优化前 vs 优化后

❌ 未优化（每次问答）：
• 发送全部文档：50,000 tokens
• 回答：500 tokens
• 总计：50,500 tokens
• 费用：$0.50（按GPT-4计价）

✅ 优化后（使用向量检索）：
• 只发送相关片段：3,000 tokens
• 回答：500 tokens
• 总计：3,500 tokens
• 费用：$0.035
• 节省：93%！

5.2 具体优化策略

1. 智能缓存

import functools
from datetime import datetime, timedelta

# 缓存装饰器
def cache_result(expire_minutes=60):
    cache = {}
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(question, *args, **kwargs):
            # 检查缓存
            if question in cache:
                result, timestamp = cache[question]
                if datetime.now() - timestamp < timedelta(minutes=expire_minutes):
                    print("✓ 使用缓存结果")
                    return result
            
            # 调用API
            result = func(question, *args, **kwargs)
            cache[question] = (result, datetime.now())
            return result
        
        return wrapper
    return decorator

# 使用
@cache_result(expire_minutes=30)
def answer_question(question, context):
    # API调用...
    pass

2. 批量处理

def batch_questions(questions: List[str], context: str) -> List[str]:
    """
    批量处理多个问题，减少API调用次数
    """
    combined_prompt = f"""基于以下文档回答多个问题：

文档：
{context}

问题：
"""
    for i, q in enumerate(questions, 1):
        combined_prompt += f"{i}. {q}\n"
    
    # 一次API调用回答所有问题
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",  # 使用更便宜的模型
        messages=[{"role": "user", "content": combined_prompt}]
    )
    
    return parse_multiple_answers(response.choices[0].message.content)

3. 模型降级策略

def smart_model_selection(question: str, context_length: int) -> str:
    """根据问题复杂度选择模型"""
    
    # 简单问题 → 便宜模型
    simple_keywords = ['什么是', '定义', '介绍']
    if any(kw in question for kw in simple_keywords) and context_length < 2000:
        return "gpt-3.5-turbo"  # $0.0015/1K tokens
    
    # 复杂推理 → 高级模型
    return "gpt-4"  # $0.03/1K tokens

# 使用
model = smart_model_selection(question, len(context))
response = client.chat.completions.create(model=model, ...)

5.3 成本监控

import tiktoken

class CostTracker:
    def __init__(self):
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        
        # GPT-4价格（每1K tokens）
        self.input_price = 0.03
        self.output_price = 0.06
    
    def count_tokens(self, text: str, model: str = "gpt-4") -> int:
        """计算token数量"""
        encoding = tiktoken.encoding_for_model(model)
        return len(encoding.encode(text))
    
    def track(self, input_text: str, output_text: str):
        """记录使用"""
        input_tokens = self.count_tokens(input_text)
        output_tokens = self.count_tokens(output_text)
        
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
    
    def get_cost(self) -> dict:
        """计算总成本"""
        input_cost = (self.total_input_tokens / 1000) * self.input_price
        output_cost = (self.total_output_tokens / 1000) * self.output_price
        
        return {
            'input_tokens': self.total_input_tokens,
            'output_tokens': self.total_output_tokens,
            'input_cost': f"${input_cost:.4f}",
            'output_cost': f"${output_cost:.4f}",
            'total_cost': f"${input_cost + output_cost:.4f}"
        }

# 使用
tracker = CostTracker()
tracker.track(prompt, response)
print(tracker.get_cost())
# {'total_cost': '$0.0234'}

---

六、部署与测试 🚀

6.1 本地测试

# 运行应用
streamlit run app.py

# 测试问题列表
测试1：上传技术文档，问"主要技术栈是什么？"
测试2：上传PDF论文，问"研究方法有哪些？"
测试3：问文档中不存在的内容，检查AI是否会说"不知道"

6.2 Docker部署

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8501

CMD ["streamlit", "run", "app.py", "--server.port=8501"]

# 构建镜像
docker build -t doc-qa-app .

# 运行容器
docker run -p 8501:8501 --env-file .env doc-qa-app

6.3 云端部署（Streamlit Cloud）

# 1. 推送代码到GitHub
git init
git add .
git commit -m "Initial commit"
git push

# 2. 访问 share.streamlit.io
# 3. 连接GitHub仓库
# 4. 配置环境变量（OPENAI_API_KEY）
# 5. 点击Deploy

---

七、常见问题解决 ❓

7.1 API调用错误

问题1：Rate Limit Error

错误信息：
"Rate limit reached for requests"

解决方案：
from tenacity import retry, wait_exponential, stop_after_attempt

@retry(wait=wait_exponential(min=1, max=60), stop=stop_after_attempt(3))
def call_api_with_retry():
    return client.chat.completions.create(...)

# 自动重试，指数退避

问题2：Token超限

错误信息：
"This model's maximum context length is 8192 tokens"

解决方案：
# 检查并截断输入
def truncate_text(text: str, max_tokens: int = 6000) -> str:
    encoding = tiktoken.get_encoding("cl100k_base")
    tokens = encoding.encode(text)
    
    if len(tokens) > max_tokens:
        tokens = tokens[:max_tokens]
        text = encoding.decode(tokens)
    
    return text

7.2 回答质量问题

问题：AI回答不准确

优化提示词：

# ❌ 模糊提示
prompt = f"根据文档回答：{question}"

# ✅ 明确提示
prompt = f"""你是文档分析专家。请严格基于以下文档内容回答问题。

规则：
1. 只使用文档中的信息
2. 如果文档没有提到，明确说"文档中未提及"
3. 引用具体段落支持你的回答
4. 保持客观，不要推测

文档：
{context}

问题：{question}

回答："""

7.3 性能优化

问题：响应太慢

优化方案：

# 1. 使用流式输出
def stream_answer(question, context):
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=[...],
        stream=True  # 启用流式
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

# 2. 异步处理
import asyncio
from openai import AsyncOpenAI

async def async_answer(question, context):
    client = AsyncOpenAI()
    response = await client.chat.completions.create(...)
    return response

---

八、总结与扩展 🎓

8.1 核心要点回顾

✓ 向量检索是关键：降低90%成本
✓ 提示词工程很重要：决定回答质量
✓ 成本监控不可少：避免预算超支
✓ 错误处理要完善：提升用户体验

8.2 进阶扩展方向

1. 多模态支持
   → 添加图片、表格识别
   → 使用GPT-4 Vision API

2. 多轮对话
   → 保存对话上下文
   → 实现追问功能

3. 知识图谱
   → 构建实体关系
   → 更智能的信息检索

4. 多语言支持
   → 自动语言检测
   → 多语言文档处理

8.3 学习资源

官方文档：
• OpenAI API: https://platform.openai.com/docs
• LangChain: https://python.langchain.com

推荐阅读：
• 《Building LLM Applications》
• OpenAI Cookbook (GitHub)

实践项目：
• 构建个人知识库
• 智能客服机器人
• 代码助手

觉得有帮助？点赞👍 收藏⭐ 关注🔔 支持作者！