1. 环境准备与工具安装要构建基于LangChain和通义千问的知识库问答API首先需要搭建开发环境。这里推荐使用Python 3.8及以上版本因为LangChain和FastAPI对较新的Python版本支持更好。安装核心依赖包很简单只需要几条pip命令pip install langchain fastapi uvicorn dashscope chromadb这些包各自发挥着重要作用langchain提供文档处理、检索和与大模型交互的能力fastapi构建高性能Web API的框架uvicorn用于运行FastAPI应用的ASGI服务器dashscope通义千问的Python SDKchromadb轻量级向量数据库用于存储文档向量我建议创建一个独立的虚拟环境来管理这些依赖避免与其他项目产生冲突。可以使用以下命令创建并激活虚拟环境python -m venv venv source venv/bin/activate # Linux/Mac venv\Scripts\activate # Windows在实际项目中我遇到过因为依赖版本不匹配导致的各种奇怪问题。为了避免这种情况建议使用requirements.txt文件固定依赖版本。创建一个包含以下内容的requirements.txt文件langchain0.0.346 fastapi0.104.1 uvicorn0.23.2 dashscope1.14.0 chromadb0.4.15然后通过pip install -r requirements.txt安装所有依赖。这种方式能确保在不同环境中的一致性特别适合团队协作场景。2. 文档预处理与向量化存储知识库问答系统的核心在于如何有效地存储和检索文档知识。我们使用LangChain提供的工具链来完成这个流程。首先准备一个结构化的问答对文档如qa.txt格式如下问题弦丝画制作的活动时长是多少 答案弦丝画制作活动时长是2-3小时。 问题用弦丝画制作福字是多少元一位 答案用弦丝画制作福字、旺字等单字款价格120元/人。文档加载和分割的代码如下from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader TextLoader(qa.txt) data loader.load() text_splitter RecursiveCharacterTextSplitter( chunk_size50, chunk_overlap0, separators[\n问题, \n答案] ) all_splits text_splitter.split_documents(data)这里有几个关键点需要注意chunk_size控制每个文本块的最大长度根据文档特点调整chunk_overlap设置文本块之间的重叠部分有助于保持上下文separators自定义分隔符确保问题和答案被正确分割接下来我们需要将这些文本块转换为向量并存储到向量数据库中from langchain.embeddings.dashscope import DashScopeEmbeddings from langchain_community.vectorstores import Chroma embeddings DashScopeEmbeddings( modeltext-embedding-v1, ) vectorstore Chroma.from_documents( documentsall_splits, embeddingembeddings, persist_directory./chroma_db )在实际项目中我发现向量数据库的持久化persist_directory非常重要。这样即使重启服务也不需要重新生成向量大大提高了开发效率。3. 构建检索增强生成(RAG)链检索增强生成(Retrieval-Augmented Generation)是LangChain的核心能力之一它结合了检索和生成两种技术能够基于知识库内容生成更准确的回答。配置通义千问大模型和RAG链的代码如下from langchain_community.llms import Tongyi from langchain.chains import RetrievalQA import os os.environ[DASHSCOPE_API_KEY] your-api-key-here llm Tongyi( model_nameqwen-turbo, temperature0.1, top_p0.9 ) qa_chain RetrievalQA.from_chain_type( llm, retrievervectorstore.as_retriever(), return_source_documentsTrue )这里有几个关键参数需要理解temperature控制生成文本的随机性值越低结果越确定top_p核采样参数影响生成文本的多样性return_source_documents设置为True可以返回检索到的原始文档便于调试我在实际使用中发现调整temperature和top_p参数对回答质量影响很大。对于知识库问答这种需要准确性的场景建议将temperature设为较低值(0.1-0.3)top_p设为0.9左右。测试RAG链的工作是否正常result qa_chain({query: 弦丝画制作的活动时长是多少}) print(result[result])如果一切正常你应该能看到系统返回了qa.txt中对应的答案。如果遇到问题可以检查API密钥是否正确设置向量数据库是否成功创建文档分割是否符合预期4. 使用FastAPI构建问答接口现在我们已经有了核心的问答能力接下来用FastAPI将其封装成Web服务。FastAPI以其高性能和易用性著称特别适合构建这类API服务。首先定义基本的FastAPI应用和请求模型from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class QuestionRequest(BaseModel): question: str然后创建API端点处理问答请求app.post(/answer) async def get_answer(request: QuestionRequest): try: result qa_chain({query: request.question}) return { answer: result[result], sources: [doc.page_content for doc in result[source_documents]] } except Exception as e: raise HTTPException(status_code500, detailstr(e))这个端点做了以下几件事接收包含question字段的JSON请求调用之前构建的qa_chain获取答案返回答案和来源文档处理可能出现的异常为了提高API的可用性我们可以添加一些额外的功能app.get(/health) async def health_check(): return {status: healthy} app.get(/examples) async def get_example_questions(): examples [] for doc in all_splits: if 问题 in doc.page_content: examples.append(doc.page_content.replace(问题, ).strip()) return {examples: examples[:5]}/health端点用于服务健康检查/examples端点返回知识库中的示例问题方便API使用者了解系统能力。最后添加启动代码if __name__ __main__: import uvicorn uvicorn.run( appmain:app, host0.0.0.0, port8080, reloadTrue, workers4 )这里有几个值得注意的参数host0.0.0.0允许从外部访问服务reloadTrue开发模式下自动重载workers4设置工作进程数提高并发能力5. 接口测试与部署完成代码编写后我们可以使用多种方式测试API。最直接的方法是使用FastAPI自带的/docs界面它提供了交互式API文档。启动服务后访问http://127.0.0.1:8080/docs你会看到一个Swagger UI界面。在这里可以尝试/health端点检查服务状态使用/examples获取示例问题通过/answer端点提交自定义问题对于生产环境部署建议使用以下uvicorn命令uvicorn main:app --host 0.0.0.0 --port 8080 --workers 4如果需要更高的性能可以考虑使用Gunicorn作为进程管理器配置Nginx作为反向代理启用HTTPS加密实现负载均衡一个基本的Nginx配置示例如下server { listen 80; server_name yourdomain.com; location / { proxy_pass http://127.0.0.1:8080; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }在实际部署中我遇到过几个常见问题跨域问题可以通过FastAPI的CORSMiddleware解决超时设置对于复杂查询需要调整默认超时时间API限流使用FastAPI的中间件实现基本限流6. 性能优化与扩展随着知识库规模的增长系统性能可能会成为瓶颈。以下是一些优化建议向量检索优化调整chunk_size和chunk_overlap参数找到最佳平衡点使用更高效的向量索引如HNSW考虑分片策略将大知识库分成多个向量库缓存策略from fastapi_cache import FastAPICache from fastapi_cache.backends.redis import RedisBackend from fastapi_cache.decorator import cache app.post(/answer) cache(expire300) # 缓存5分钟 async def get_answer(request: QuestionRequest): ...异步处理 对于耗时较长的查询可以考虑使用Celery等任务队列实现异步处理from celery import Celery celery_app Celery(tasks, brokerredis://localhost:6379/0) celery_app.task def async_qa(query): return qa_chain({query: query}) app.post(/async_answer) async def get_async_answer(request: QuestionRequest): task async_qa.delay(request.question) return {task_id: task.id}监控与日志 集成Prometheus和Grafana监控关键指标请求响应时间错误率向量检索耗时大模型生成耗时from prometheus_fastapi_instrumentator import Instrumentator Instrumentator().instrument(app).expose(app)7. 实际应用中的经验分享在多个项目中实施这种架构后我总结了一些实用技巧知识库维护定期更新qa.txt文件保持知识新鲜度实现一个管理接口支持动态更新知识库对用户反馈的错误答案进行记录和分析错误处理 增强API的错误处理能力app.post(/answer) async def get_answer(request: QuestionRequest): if not request.question.strip(): raise HTTPException(status_code400, detail问题不能为空) if len(request.question) 200: raise HTTPException(status_code400, detail问题过长) try: result qa_chain({query: request.question}) if not result[result]: raise HTTPException(status_code404, detail未找到相关答案) return {answer: result[result]} except Exception as e: logger.error(f问答出错: {str(e)}) raise HTTPException(status_code500, detail系统处理问题出错)安全考虑实现API密钥认证对用户输入进行安全检查限制敏感信息的返回from fastapi.security import APIKeyHeader api_key_header APIKeyHeader(nameX-API-KEY) app.post(/answer) async def get_answer( request: QuestionRequest, api_key: str Depends(api_key_header) ): if api_key ! your-secret-key: raise HTTPException(status_code403, detail无效的API密钥) ...效果评估 建立评估机制监控问答质量记录用户反馈定期抽样检查使用自动化测试用例test_cases [ {question: 弦丝画制作的活动时长是多少, expected: 2-3小时}, {question: 福字制作价格是多少, expected: 120元/人} ] for case in test_cases: result qa_chain({query: case[question]}) assert case[expected] in result[result]