迭代改进
收集反馈
1. 添加反馈机制
在应用中添加反馈按钮：
python
# 在 app.py 中添加
with st.sidebar:
st.subheader("📝 反馈")
feedback = st.text_area(
"这个回答有用吗？有什么建议？",
placeholder="告诉我们你的想法..."
)
if st.button("提交反馈"):
# 保存到文件或数据库
with open("feedback.txt", "a") as f:
f.write(f"\n{feedback}\n")
st.success("感谢反馈！")2. 查看使用日志
Streamlit 会记录所有请求，可以在后台查看：
- 哪些问题被问得最多
- 用户停留时间
- 错误率
常见改进方向
改进 1：提高准确率
问题：用户反馈答案不准确
解决方案：
- 优化检索策略
python
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
# 使用 LLM 压缩检索结果
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever
)
# 使用压缩后的检索器
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
retriever=compression_retriever
)- 调整 chunk 参数
实验不同的参数组合：
python
# 测试不同的 chunk_size
for size in [300, 500, 800, 1000]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=size,
chunk_overlap=50
)
# 测试效果- 使用更好的 embedding 模型
python
from langchain_community.embeddings import HuggingFaceEmbeddings
# 使用开源的中文 embedding
embeddings = HuggingFaceEmbeddings(
model_name="shibing624/text2vec-base-chinese"
)改进 2：提升响应速度
问题：用户反馈太慢
解决方案：
- 缓存常见问题
python
import hashlib
@st.cache_data
def get_cached_answer(question, vectorstore):
# 对问题生成哈希作为缓存键
cache_key = hashlib.md5(question.encode()).hexdigest()
# 检查缓存
if cache_key in st.session_state:
return st.session_state[cache_key]
# 生成答案
result = answer_question(question, vectorstore)
# 保存到缓存
st.session_state[cache_key] = result
return result- 使用更快的模型
python
# 在配置中添加
FAST_LLM = "qwen-turbo" # 快速模式
ACCURATE_LLM = "qwen-plus" # 准确模式
# 根据问题复杂度选择
if len(question) < 50:
llm = ChatTongyi(model=FAST_LLM)
else:
llm = ChatTongyi(model=ACCURATE_LLM)- 并行处理
python
from concurrent.futures import ThreadPoolExecutor
def parallel_retrieve(vectorstore, questions):
"""并行检索多个问题"""
with ThreadPoolExecutor(max_workers=3) as executor:
results = executor.map(
lambda q: retrieve_documents(vectorstore, q),
questions
)
return list(results)改进 3：增强功能
添加来源高亮
python
def highlight_sources(answer, sources):
"""在答案中高亮显示来源"""
for i, source in enumerate(sources):
# 在答案后添加来源链接
answer += f"\n\n[{i+1}] {source['metadata']['source']}"
return answer支持多语言
python
from langdetect import detect
def detect_language(text):
"""检测文本语言"""
return detect(text)
language = detect_language(question)
if language == 'zh':
prompt_template = "你是一个中文问答助手..."
else:
prompt_template = "You are an English QA assistant..."添加相关推荐
python
def get_related_questions(question, vectorstore):
"""推荐相关问题"""
# 使用向量相似度找相似问题
similar_questions = [
"什么是机器学习？",
"深度学习和机器学习的区别",
"如何开始学习 AI？"
]
return similar_questions[:3]
# 在界面中显示
related = get_related_questions(question, st.session_state.vectorstore)
st.write("💡 你可能还想问：")
for q in related:
if st.button(q):
st.session_state.current_question = q
st.rerun()改进 4：优化用户体验
添加示例问题
python
example_questions = [
"这个文档主要讲了什么？",
"有哪些关键要点？",
"有什么具体例子吗？"
]
st.write("👆 点击示例问题快速开始：")
cols = st.columns(len(example_questions))
for i, q in enumerate(example_questions):
with cols[i]:
if st.button(q):
st.session_state.example_question = q
st.rerun()改进错误提示
python
def safe_answer_question(question, vectorstore):
"""带错误处理的问答"""
try:
result = answer_question(question, vectorstore)
return result
except Exception as e:
# 友好的错误提示
error_messages = {
"api_key": "API 配置错误，请检查设置",
"network": "网络连接失败，请稍后重试",
"timeout": "响应超时，请简化问题"
}
for key, msg in error_messages.items():
if key in str(e).lower():
return {
"answer": f"❌ {msg}",
"sources": []
}
# 未知错误
return {
"answer": "❌ 处理失败，请换个问题试试",
"sources": []
}数据分析
收集使用数据
python
import json
from datetime import datetime
def log_question(question, answer, user_feedback=None):
"""记录问题和答案"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"question": question,
"answer": answer[:100] + "...", # 只保存前 100 字
"feedback": user_feedback
}
with open("usage_logs.jsonl", "a") as f:
f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")分析数据
创建 analyze_logs.py：
python
import json
from collections import Counter
def analyze_logs():
"""分析使用日志"""
questions = []
feedback_scores = []
with open("usage_logs.jsonl", "r") as f:
for line in f:
entry = json.loads(line)
questions.append(entry["question"])
# 统计最常见的问题
top_questions = Counter(questions).most_common(10)
print("最常见的问题：")
for q, count in top_questions:
print(f" {count}x: {q}")
if __name__ == "__main__":
analyze_logs()版本迭代建议
v1.0（当前版本）
- 基本问答功能
- 单文档支持
- 简单界面
v1.1（短期优化）
- [ ] 优化准确率
- [ ] 提升响应速度
- [ ] 添加示例问题
- [ ] 改进错误提示
v1.2（功能增强）
- [ ] 多文档支持
- [ ] 对话历史
- [ ] 导出对话
- [ ] 反馈系统
v2.0（重大升级）
- [ ] 用户系统
- [ ] 多语言支持
- [ ] 高级检索模式
- [ ] API 接口
持续学习
关注最新技术
- LangChain 更新：每月查看新功能
- 新模型发布：测试新的 LLM
- 向量数据库：关注性能改进
社区资源
下一个项目
完成了这个项目，你已经掌握了：
- RAG 的基本原理
- 向量数据库的使用
- LLM API 调用
- 简单的应用部署
接下来可以挑战：
- 项目二：多 Agent 研究助手 - 学习 Agent 协作
- 项目四：跨平台内容分发工具 - 学习 API 集成
项目回顾
恭喜！你完成了第一个 AI Agent 项目。
你学会了
- ✅ RAG 的完整流程
- ✅ 向量数据库的使用
- ✅ LLM API 集成
- ✅ Streamlit 快速开发
- ✅ 应用部署上线
继续前进
AI Agent 开发才刚刚开始。继续做更多项目，深入理解 Agent 的设计和实现。