1. 项目概述:20分钟快速构建LangChain智能体
去年我在一个企业知识库项目中首次接触LangChain时,花了整整三天才搞明白如何搭建一个基础问答系统。现在回头看,其实用LangChain开发一个功能完善的智能体(Agent),20分钟真的足够了。本文将以一个文档问答Agent为例,带你快速掌握LangChain Agent开发的核心技巧。
这个智能体的核心功能是:基于LangChain官方文档构建知识库,当用户提出技术问题时,它能自动检索相关文档片段,通过子智能体(Subagent)并行分析多个文档片段,最终合成专业回答。整个过程涉及工具调用、任务分解、并行处理等典型Agent能力,是学习LangChain的理想入门项目。
2. 环境准备与依赖安装
2.1 基础环境配置
建议使用Python 3.9+环境,我习惯用conda创建独立环境:
conda create -n langchain-agent python=3.9 conda activate langchain-agent2.2 核心依赖安装
这些是开发LangChain Agent必需的库:
pip install langchain-core langchain-openai deepagents pip install langchain-text-splitters requests注意:如果遇到依赖冲突,可以先安装langchain-core再安装其他组件。我遇到过langchain-community与新版langchain-core不兼容的情况,指定版本号可以解决:
pip install langchain-core==0.1.0
2.3 模型API配置
示例使用OpenAI的GPT-4模型,你需要准备:
- OpenAI API密钥
- 在环境变量中设置:
import os os.environ["OPENAI_API_KEY"] = "你的API密钥"如果无法访问OpenAI,也可以使用本地模型如Ollama:
from langchain_community.llms import Ollama llm = Ollama(model="llama3")3. Agent核心架构设计
3.1 整体工作流程
我们的文档问答Agent采用经典RAG架构:
- 检索(Retrieve):用向量数据库搜索相关文档片段
- 分析(Analyze):子智能体并行处理文档片段
- 生成(Generate):主智能体综合所有分析结果生成最终回答
3.2 关键技术组件
| 组件类型 | 实现方案 | 作用 |
|---|---|---|
| 文本分割器 | RecursiveCharacterTextSplitter | 将长文档切分为可处理的片段 |
| 向量数据库 | InMemoryVectorStore | 存储和检索文档片段 |
| 主智能体 | create_deep_agent | 协调整个问答流程 |
| 子智能体 | chunk-analyst | 分析单个文档片段 |
4. 文档处理与索引构建
4.1 文档加载实现
我们从LangChain官方文档网站加载技术文档:
from langchain_core.documents import Document import requests def load_docs(doc_urls): docs = [] for url in doc_urls: try: resp = requests.get(url, timeout=10) docs.append(Document( page_content=resp.text, metadata={"source": url} )) except Exception as e: print(f"加载失败 {url}: {str(e)}") return docs doc_urls = [ "https://docs.langchain.com/docs/agents", "https://docs.langchain.com/docs/retrieval" ] documents = load_docs(doc_urls)4.2 文本分块策略
文档分块是RAG系统的关键,我的经验是:
- 技术文档适合800-1200字符的块大小
- 设置200字符的重叠区域防止关键信息被切断
from langchain_text_splitters import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, separators=["\n\n", "\n", "。", "!", "?"] ) splits = text_splitter.split_documents(documents)4.3 向量存储配置
使用内存型向量数据库简化示例:
from langchain_openai import OpenAIEmbeddings from langchain_core.vectorstores import InMemoryVectorStore embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vector_store = InMemoryVectorStore.from_documents( documents=splits, embedding=embeddings )生产环境建议使用Pinecone或Chroma等专业向量数据库。我在实际项目中发现,当文档超过1万条时,内存数据库的性能会显著下降。
5. 智能体核心功能实现
5.1 搜索工具开发
这是Agent与向量数据库交互的关键工具:
from deepagents.backends import StateBackend from langchain.tools import tool import uuid backend = StateBackend() @tool def search_docs(query: str) -> list: """检索与问题相关的文档片段""" results = vector_store.similarity_search(query, k=3) # 将结果保存到临时文件供子智能体读取 batch_id = uuid.uuid4().hex[:6] saved_paths = [] for i, doc in enumerate(results): path = f"/retrieved/{batch_id}/doc_{i}.txt" content = f"来源: {doc.metadata['source']}\n内容: {doc.page_content}" backend.upload_file(path, content.encode('utf-8')) saved_paths.append(path) return saved_paths5.2 子智能体配置
子智能体专门分析单个文档片段:
chunk_analyst = { "name": "doc-analyst", "description": "分析单个文档片段", "system_prompt": """你是一个技术文档分析专家。请根据用户问题,从指定文档片段中提取关键信息。 要求: 1. 只基于提供的文档内容回答 2. 保持回答简洁专业 3. 必须注明文档来源""" }5.3 主智能体创建
整合所有组件创建主智能体:
from deepagents import create_deep_agent from langchain.chat_models import ChatOpenAI agent = create_deep_agent( model=ChatOpenAI(model="gpt-4"), tools=[search_docs], backend=backend, system_prompt="""你是LangChain技术专家,请按以下流程回答问题: 1. 用search_docs工具检索相关文档 2. 为每个文档片段创建分析任务 3. 综合所有分析结果生成最终回答""", subagents=[chunk_analyst] )6. 智能体测试与优化
6.1 基础功能测试
question = "LangChain Agent如何调用工具?" result = agent.invoke({"messages": [{"content": question}]}) print(result["messages"][-1]["content"])预期输出应包含:
- 工具调用的基本方法
- 相关代码示例
- 文档来源链接
6.2 性能优化技巧
通过实际测试发现的优化点:
- 并发控制:设置合理的子智能体并发数(通常3-5个)
agent = create_deep_agent( ... max_concurrent_subagents=3 )- 结果缓存:对常见问题实现答案缓存
from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())- 检索优化:调整相似度搜索参数
results = vector_store.similarity_search( query, k=5, # 返回结果数 score_threshold=0.7 # 相似度阈值 )7. 生产环境部署建议
7.1 安全注意事项
- 输入过滤:防止Prompt注入攻击
import re def sanitize_input(text): return re.sub(r"[^\w\s.,?]", "", text)[:500]- 输出审查:确保回答的准确性
def validate_answer(answer): if "我不知道" in answer: raise ValueError("回答不完整") if not any(url in answer for url in doc_urls): print("警告:回答未引用文档")7.2 监控与日志
建议添加:
from langsmith import Client client = Client() client.create_project("DocQA-Agent") def log_interaction(question, answer): client.create_example( inputs={"question": question}, outputs={"answer": answer}, project_name="DocQA-Agent" )8. 常见问题排查
8.1 检索相关问题
问题:检索结果不相关解决:
- 检查嵌入模型是否匹配(text-embedding-3-small最适合英文)
- 调整分块策略,技术文档可能需要更小的块(500-800字符)
8.2 子智能体问题
问题:子智能体不返回结果检查:
- 确认文件路径是否正确
- 检查子智能体的system_prompt是否明确
- 查看LangSmith跟踪日志
8.3 性能问题
问题:响应速度慢优化:
- 限制子智能体响应长度
chunk_analyst["system_prompt"] += "\n请将回答限制在200字以内"- 使用更快的模型如gpt-3.5-turbo
9. 项目扩展方向
这个基础Agent可以进一步扩展:
- 多模态支持:处理图片、PDF等格式文档
from langchain.document_loaders import PyPDFLoader loader = PyPDFLoader("manual.pdf") pages = loader.load()- 对话记忆:添加聊天历史功能
from langchain.memory import ConversationBufferMemory memory = ConversationBufferMemory() agent = create_deep_agent( ..., memory=memory )- 自动验证:添加事实核查子智能体
fact_checker = { "name": "fact-checker", "description": "验证回答的准确性", "system_prompt": "检查技术回答是否符合官方文档..." }我在实际项目中发现,开发一个基础Agent确实可以在20分钟内完成,但要打造生产级应用,还需要在异常处理、性能优化等方面投入更多时间。建议先从这个小项目开始,逐步添加更复杂的功能。