LangChain与LangGraph：AI工作流开发实战指南

1. 项目概述LangChain和LangGraph是当前AI应用开发领域最热门的两大框架工具组合。它们让开发者能够快速构建基于大语言模型(LLM)的复杂工作流就像搭积木一样把不同的AI能力连接起来。我在实际项目中用这套技术栈开发过客服对话系统、智能文档分析工具等多个应用今天就来分享从零开始的使用指南。对于刚接触AI开发的初学者来说这两个工具最大的价值在于用声明式方法定义AI处理流程内置常见模式如问答、摘要、分类的预制组件可视化调试复杂的多步骤工作流2. 核心概念解析2.1 LangChain基础架构LangChain的核心是链(Chain)的概念。一个典型的链包含这些要素from langchain.chains import LLMChain from langchain.prompts import PromptTemplate prompt PromptTemplate( input_variables[product], template给这款{product}写一段电商商品描述要求突出卖点 ) chain LLMChain(llmllm, promptprompt)这里构建的是一条商品文案生成链。实际开发中常见的链类型包括数据提取链(Extraction)问答链(QA)摘要链(Summarization)2.2 LangGraph的核心优势LangGraph在LangChain基础上增加了工作流编排能力。它的两大特色功能有状态执行可以记住对话历史、中间结果条件分支根据AI输出决定后续步骤比如这个客服对话流程from langgraph.graph import Graph workflow Graph() workflow.add_node(greet, greet_user) workflow.add_node(query, handle_query) workflow.add_edge(greet, query)3. 环境搭建指南3.1 安装准备推荐使用Python 3.10环境通过pip安装核心包pip install langchain langgraph openai对于国内用户建议配置清华镜像源加速下载pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple3.2 API密钥配置主流LLM服务的配置方法# OpenAI import os os.environ[OPENAI_API_KEY] sk-xxx # 国内平台示例 os.environ[DASHSCOPE_API_KEY] sk-xxx重要提示永远不要将API密钥提交到Git仓库建议使用环境变量或密钥管理工具4. 第一个AI工作流实战4.1 智能邮件分类器我们构建一个能自动分类并回复邮件的系统from langchain.chains import TransformChain def classify_email(inputs): content inputs[email_content] # 调用LLM判断邮件类型 return {category: 投诉 if 不满意 in content else 咨询} classification_chain TransformChain( input_variables[email_content], output_variables[category], transformclassify_email )4.2 集成到LangGraph将多个链组合成完整工作流workflow Graph() workflow.add_node(classify, classification_chain) workflow.add_node(respond, response_chain) workflow.add_conditional_edges( classify, lambda x: complaint if x[category]投诉 else normal, {complaint: escalate, normal: respond} )5. 高级应用技巧5.1 记忆机制实现让AI记住对话历史的关键代码from langchain.memory import ConversationBufferMemory memory ConversationBufferMemory() chain ConversationChain( llmllm, memorymemory )5.2 文件处理实战解析PDF文档的完整流程from langchain.document_loaders import PyPDFLoader loader PyPDFLoader(report.pdf) pages loader.load_and_split() text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) docs text_splitter.split_documents(pages)6. 性能优化方案6.1 缓存策略使用SQLite缓存LLM响应提升速度from langchain.cache import SQLiteCache import langchain langchain.llm_cache SQLiteCache(database_path.langchain.db)6.2 异步处理提高吞吐量的异步调用方式async def batch_process(texts): chain LLMChain(llmllm, promptprompt) return await chain.arun(texts)7. 常见问题排查7.1 超时错误处理当遇到API超时时建议这样重试from tenacity import retry, stop_after_attempt retry(stopstop_after_attempt(3)) def safe_call(prompt): return llm(prompt)7.2 中文优化技巧提升中文处理效果的方法在prompt中明确指定用中文回答对长文本先分块再处理使用本地化模型如ChatGLMprompt 请用简体中文回答以下问题 问题{question} 答案8. 项目扩展思路8.1 接入企业微信通过自定义Tool实现from langchain.tools import tool tool def send_wecom_msg(content: str): 发送企业微信消息 # 实现企业微信API调用 return 消息已发送8.2 可视化监控使用Prometheus监控调用指标from prometheus_client import Counter api_calls Counter(langchain_api_calls, API调用统计) def wrapped_llm(prompt): api_calls.inc() return llm(prompt)我在实际项目中发现将LangChain与FastAPI结合可以快速构建AI微服务。一个典型的部署架构包含前端Vue.js交互界面后端FastAPI处理请求AI层LangChain工作流存储PostgreSQL向量数据库这种架构下平均响应时间可以控制在800ms以内适合大多数企业级应用场景。对于需要更高并发的场景可以考虑使用Celery进行任务队列管理。