用langchain写个示例代码调用它的特色功能-平芜编程栈

当然可以！下面是一个完整的 Python 实例，使用 LangChain（v0.2+ / v1.x 兼容）演示其六大核心特色功能：

✅ 模块化组件（Prompt + LLM + OutputParser）
✅ 外部数据集成（RAG：加载本地文档 + 向量检索）
✅ 记忆（Memory：多轮对话上下文）
✅ 链（Chain：声明式 LCEL 流水线）
✅ 代理（Agent：调用工具执行动作）
✅ 回调（Callback：打印日志）

💡 假设你已安装必要依赖：

pip install langchain langchain-openai langchain-community faiss-cpu pypdf

🧪 完整代码示例：带记忆、知识库和工具的智能客服

import os from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain.chains import create_history_aware_retriever, create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain from langchain_core.messages import HumanMessage, AIMessage from langchain.memory import ConversationBufferMemory from langchain.agents import Tool, initialize_agent, AgentType from langchain.callbacks import StdOutCallbackHandler # === 1. 设置大模型（Model I/O）=== llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.3, api_key=os.getenv("OPENAI_API_KEY") # 请设置环境变量 ) # === 2. 加载外部知识（RAG 数据连接）=== # 假设当前目录有 `faq.pdf`（可替换为任意 PDF/文本） loader = PyPDFLoader("faq.pdf") docs = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) splits = text_splitter.split_documents(docs) vectorstore = FAISS.from_documents(splits, OpenAIEmbeddings()) retriever = vectorstore.as_retriever() # === 3. 构建带记忆的检索问答链（Chain + Memory）=== prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个客服助手。请基于以下上下文回答问题：\n{context}"), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}") ]) document_chain = create_stuff_documents_chain(llm, prompt) retriever_chain = create_history_aware_retriever(llm, retriever, ChatPromptTemplate.from_messages([ MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), ("human", "根据对话历史，生成一个搜索查询以查找相关信息。") ]) ) qa_chain = create_retrieval_chain(retriever_chain, document_chain) # === 4. 定义工具（Tools for Agent）=== def send_email(to: str, content: str) -> str: # 模拟发邮件（实际可调用 SMTP 或 API） return f"✅ 邮件已发送至 {to}，内容：{content[:30]}..." email_tool = Tool( name="send_email", func=send_email, description="当用户要求发送邮件时使用，参数：to（邮箱地址），content（邮件内容）" ) # === 5. 初始化带工具的 Agent（Agents）=== agent = initialize_agent( tools=[email_tool], llm=llm, agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, verbose=True, memory=ConversationBufferMemory(memory_key="chat_history", return_messages=True), handle_parsing_errors=True ) # === 6. 回调处理器（Callbacks）=== handler = StdOutCallbackHandler() # === 7. 模拟用户交互 === chat_history = [] def ask_question(user_input: str): global chat_history # 先尝试用 RAG 知识库回答 try: response = qa_chain.invoke({ "input": user_input, "chat_history": chat_history }) answer = response["answer"] chat_history.extend([HumanMessage(content=user_input), AIMessage(content=answer)]) print(f"🤖 (知识库): {answer}\n") except Exception as e: answer = None # 如果知识库无法回答，交给 Agent 决策是否调用工具 if not answer or "不知道" in answer or "无法回答" in answer: print("🔄 转交 Agent 处理...") agent_response = agent.run(user_input, callbacks=[handler]) chat_history.append(HumanMessage(content=user_input)) chat_history.append(AIMessage(content=agent_response)) print(f"🤖 (Agent): {agent_response}\n") # === 8. 测试对话 === if __name__ == "__main__": # 示例：先问知识库问题，再要求发邮件 ask_question("你们的退货政策是什么？") # 从 faq.pdf 中检索 ask_question("请把刚才的回答发到 user@example.com") # 触发 send_email 工具

🔍 功能亮点说明：

表格

特性	代码体现
模块化	`ChatPromptTemplate`+`StrOutputParser`+`FAISS`各司其职
RAG	用`PyPDFLoader`+`FAISS`构建私有知识库
记忆	`chat_history`传递上下文，支持多轮问答
链（LCEL）	`create_retrieval_chain`是典型 Chain 编排
Agent + Tools	`initialize_agent`+ 自定义`send_email`工具
回调	`StdOutCallbackHandler()`打印执行过程