AI Agent基础|LangChain框架入门

1. LangChain框架概述

LangChain是一个用于开发由语言模型驱动的应用程序的开源框架。它提供了一套工具和组件，使开发者能够轻松构建复杂的语言模型应用。

LangChain的核心价值在于：

• 提供了与语言模型交互的标准接口
• 支持多种语言模型提供商（OpenAI、Anthropic、Hugging Face等）
• 提供了丰富的工具生态系统
• 支持复杂的链式调用和代理模式

2. LangChain核心组件

2.1 LLMs（Large Language Models）

LLMs是LangChain的核心组件之一，提供了与各种语言模型交互的统一接口。

2.2 Prompt Templates

Prompt Templates用于管理提示语的结构化创建，支持变量替换和模板复用。

2.3 Chains

Chains允许将多个组件链接在一起，形成复杂的工作流。

2.4 Agents

Agents能够根据需要调用工具来完成任务，具有决策能力。

2.5 Tools

Tools是Agent可以调用的功能，可以是API、函数或其他服务。

2.6 Memory

Memory组件用于在对话中保持上下文信息。

2.7 Indexes

Indexes用于处理和检索外部数据源。

3. LangChain的应用场景

• 聊天机器人
• 文档问答系统
• 数据分析助手
• 自动化工具
• 内容生成

设计思路

使用LangChain构建应用的典型流程：

1. 选择合适的LLM
2. 设计Prompt模板
3. 组合Chains或创建Agents
4. 集成必要的Tools
5. 添加Memory支持（如需要）
6. 测试和优化

代码实现

下面我们将实现LangChain框架的基本使用示例：

# 注意：在实际使用时需要安装langchain # pip install langchain-community langchain-openai from typing import Dict, List, Any, Optional import json import re from datetime import datetime import math class MockLLM: """ 模拟的语言模型类 在实际应用中，这里会是真实的LLM（如OpenAI GPT） """ def __init__(self, model_name: str = "mock-llm"): self.model_name = model_name self.call_count = 0 self.responses = { "hello": "Hello! I'm an AI assistant. How can I help you today?", "time": f"The current time is {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", "weather": "I don't have access to real-time weather data, but I can help you find weather information if you tell me your location.", "math": "I can help with mathematical calculations. Please provide the specific calculation you'd like me to perform.", "default": "I understand your request. Could you please provide more details so I can assist you better?" } def predict(self, prompt: str) -> str: """ 模拟LLM预测 """ self.call_count += 1 prompt_lower = prompt.lower() # 根据提示内容返回相应的模拟响应 if "hello" in prompt_lower or "hi" in prompt_lower: return self.responses["hello"] elif "time" in prompt_lower or "date" in prompt_lower: return self.responses["time"] elif "weather" in prompt_lower: return self.responses["weather"] elif any(op in prompt_lower for op in ["calculate", "compute", "math", "+", "-", "*", "/"]): return self.responses["math"] else: return self.responses["default"] def generate(self, prompts: List[str]) -> List[str]: """ 模拟批量生成 """ return [self.predict(prompt) for prompt in prompts] class PromptTemplate: """ 提示模板类 """ def __init__(self, template: str, input_variables: List[str]): self.template = template self.input_variables = input_variables def format(self, **kwargs) -> str: """ 格式化模板 """ # 验证必需的变量是否提供 missing_vars = set(self.input_variables) - set(kwargs.keys()) if missing_vars: raise ValueError(f"缺少必需的变量: {missing_vars}") # 替换变量 formatted_prompt = self.template for var, value in kwargs.items(): placeholder = "{" + var + "}" formatted_prompt = formatted_prompt.replace(placeholder, str(value)) return formatted_prompt def partial_format(self, **kwargs) -> 'PromptTemplate': """ 部分格式化模板 """ new_template = self.template new_input_vars = [var for var in self.input_variables if var not in kwargs] for var, value in kwargs.items(): if var in self.input_variables: placeholder = "{" + var + "}" new_template = new_template.replace(placeholder, str(value)) return PromptTemplate(new_template, new_input_vars) class Tool: """ 工具基类 """ def __init__(self, name: str, description: str, func): self.name = name self.description = description self.func = func def run(self, tool_input: str) -> str: """ 运行工具 """ try: return str(self.func(tool_input)) except Exception as e: return f"Error running tool {self.name}: {str(e)}" def __call__(self, tool_input: str) -> str: """ 使工具可调用 """ return self.run(tool_input) class CalculatorTool(Tool): """ 计算器工具 """ def __init__(self): def calculate(expression: str) -> str: """ 执行数学计算 支持基本运算：+, -, *, /, **, sqrt, sin, cos, tan """ try: # 安全的数学表达式计算 # 只允许特定的安全操作 allowed_names = { k: v for k, v in math.__dict__.items() if not k.startswith("__") } # 解析表达式 # 支持格式: "2 + 3", "sqrt(16)", "sin(pi/2)" result = eval(expression, {"__builtins__": {}}, allowed_names) return f"Result: {result}" except Exception as e: return f"Calculation error: {str(e)}" super().__init__( name="Calculator", description="用于执行数学计算的工具。输入应该是一个数学表达式，例如 '2 + 3' 或 'sqrt(16)'", func=calculate ) class TimeTool(Tool): """ 时间工具 """ def __init__(self): def get_current_time(input_str: str) -> str: """ 获取当前时间 """ current_time = datetime.now() if "date" in input_str.lower(): return f"Today's date is {current_time.strftime('%Y-%m-%d')}" elif "time" in input_str.lower(): return f"The current time is {current_time.strftime('%H:%M:%S')}" else: return f"Current date and time: {current_time.strftime('%Y-%m-%d %H:%M:%S')}" super().__init__( name="Time", description="获取当前日期和时间的工具。输入可以是空或包含'date'/'time'等关键词", func=get_current_time ) class Chain: """ 链类 - 将多个组件链接在一起 """ def __init__(self, steps: List[Any]): self.steps = steps def run(self, inputs: Dict[str, Any]) -> Dict[str, Any]: """ 运行链 """ results = inputs.copy() for i, step in enumerate(self.steps): if callable(step): # 如果步骤是可调用的，直接执行 results = step(results) elif hasattr(step, 'run'): # 如果步骤有run方法，调用它 results = step.run(results) else: # 其他情况，尝试调用 results = step(results) return results class SimpleQAChain(Chain): """ 简单问答链 """ def __init__(self, llm, prompt_template: PromptTemplate): self.llm = llm self.prompt_template = prompt_template super().__init__([self._qa_step]) def _qa_step(self, inputs: Dict[str, Any]) -> Dict[str, Any]: """ 问答步骤 """ question = inputs.get('question', '') # 格式化提示 prompt = self.prompt_template.format(question=question) # 调用LLM response = self.llm.predict(prompt) return { 'question': question, 'answer': response, 'prompt_used': prompt } class Agent: """ Agent类 - 能够使用工具来完成任务 """ def __init__(self, llm, tools: List[Tool], max_iterations: int = 5): self.llm = llm self.tools = {tool.name: tool for tool in tools} self.max_iterations = max_iterations def run(self, task: str) -> str: """ 运行Agent完成任务 """ iteration = 0 intermediate_steps = [] current_task = task while iteration < self.max_iterations: # 生成决策 decision_prompt = self._create_decision_prompt(current_task, intermediate_steps) decision = self.llm.predict(decision_prompt) # 解析决策 tool_name, tool_input = self._parse_decision(decision) if tool_name and tool_name in self.tools: # 执行工具 tool_result = self.tools[tool_name].run(tool_input) intermediate_steps.append({ 'tool_name': tool_name, 'tool_input': tool_input, 'result': tool_result }) # 更新任务 current_task = f"Previous result: {tool_result}\nCurrent task: {task}" else: # 如果没有找到合适的工具，返回最终答案 return decision iteration += 1 # 如果达到最大迭代次数，返回最终结果 final_prompt = f"Based on these steps: {intermediate_steps}, please provide a final answer to the original task: {task}" return self.llm.predict(final_prompt) def _create_decision_prompt(self, task: str, intermediate_steps: List[Dict]) -> str: """ 创建决策提示 """ tools_desc = "\n".join([ f"{name}: {tool.description}" for name, tool in self.tools.items() ]) if intermediate_steps: steps_str = "\n".join([ f"Step {i+1}: Used {step['tool_name']} with input '{step['tool_input']}' -> Result: {step['result']}" for i, step in enumerate(intermediate_steps) ]) prompt = f""" Task: {task} Available tools: {tools_desc} Previous steps: {steps_str} What should I do next? Please respond in the format: <thought>Think about what needs to be done</thought> <action>ToolName[input]</action> or <final_answer>Your final answer here</final_answer> """ else: prompt = f""" Task: {task} Available tools: {tools_desc} What should I do? Please respond in the format: <thought>Think about what needs to be done</thought> <action>ToolName[input]</action> or <final_answer>Your final answer here</final_answer> """ return prompt def _parse_decision(self, decision: str) -> tuple: """ 解析决策，提取工具名和输入 """ # 查找<action>标签 action_match = re.search(r'<action>(\w+)\[(.*?)\]</action>', decision) if action_match: tool_name = action_match.group(1) tool_input = action_match.group(2) return tool_name, tool_input # 查找<final_answer>标签 answer_match = re.search(r'<final_answer>(.*?)</final_answer>', decision, re.DOTALL) if answer_match: return None, answer_match.group(1).strip() # 如果没有找到标签，尝试直接解析 # 这里可以添加更复杂的解析逻辑 return None, decision class ConversationBufferMemory: """ 对话缓冲区内存 """ def __init__(self, max_history: int = 10): self.max_history = max_history self.history = [] def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]): """ 保存对话上下文 """ self.history.append({ 'inputs': inputs.copy(), 'outputs': outputs.copy(), 'timestamp': datetime.now() }) # 限制历史记录长度 if len(self.history) > self.max_history: self.history = self.history[-self.max_history:] def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]: """ 加载内存变量 """ if not self.history: return {'history': ''} history_str = '\n'.join([ f"Q: {item['inputs'].get('question', '')}\nA: {item['outputs'].get('answer', '')}" for item in self.history ]) return {'history': history_str} def clear(self): """ 清除内存 """ self.history = [] def demo_prompt_templates(): """ 演示提示模板的使用 """ print("=== 提示模板演示 ===") # 创建一个简单的问答模板 qa_template = PromptTemplate( template="Please answer the following question: {question}", input_variables=["question"] ) formatted_prompt = qa_template.format(question="What is the capital of France?") print(f"格式化后的提示: {formatted_prompt}") # 创建一个带有多变量的模板 context_template = PromptTemplate( template="Given the context: {context}\nAnswer this question: {question}\nUse the provided format: {format_instructions}", input_variables=["context", "question", "format_instructions"] ) formatted_context_prompt = context_template.format( context="The Eiffel Tower is located in Paris, France.", question="Where is the Eiffel Tower?", format_instructions="Provide the answer in the form 'The Eiffel Tower is located in [location]'." ) print(f"上下文提示: {formatted_context_prompt}") # 部分格式化 partial_template = context_template.partial_format( context="The Great Wall of China stretches across northern China.", format_instructions="Answer in exactly one sentence." ) print(f"部分格式化后剩余变量: {partial_template.input_variables}") final_prompt = partial_template.format(question="What is the Great Wall of China?") print(f"最终提示: {final_prompt}") def demo_simple_chain(): """ 演示简单链的使用 """ print("\n=== 简单链演示 ===") # 创建模拟LLM llm = MockLLM() # 创建提示模板 qa_template = PromptTemplate( template="Please answer this question briefly: {question}", input_variables=["question"] ) # 创建问答链 qa_chain = SimpleQAChain(llm, qa_template) # 运行链 result = qa_chain.run({"question": "What is the current time?"}) print(f"问题: {result['question']}") print(f"答案: {result['answer']}") print(f"使用的提示: {result['prompt_used']}") def demo_tools(): """ 演示工具的使用 """ print("\n=== 工具演示 ===") # 创建工具 calculator = CalculatorTool() time_tool = TimeTool() # 测试计算器工具 print("计算器工具测试:") calc_results = [ calculator.run("2 + 3"), calculator.run("sqrt(16)"), calculator.run("sin(3.14159/2)"), calculator.run("10 * 5 - 3") ] for expr, result in zip(["2 + 3", "sqrt(16)", "sin(3.14159/2)", "10 * 5 - 3"], calc_results): print(f" {expr} = {result}") # 测试时间工具 print("\n时间工具测试:") time_results = [ time_tool.run(""), time_tool.run("date"), time_tool.run("time") ] for input_str, result in zip(["", "date", "time"], time_results): print(f" Input: '{input_str}' -> Output: {result}") def demo_agent(): """ 演示Agent的使用 """ print("\n=== Agent演示 ===") # 创建LLM和工具 llm = MockLLM() tools = [CalculatorTool(), TimeTool()] # 创建Agent agent = Agent(llm, tools, max_iterations=3) # 测试Agent tasks = [ "What is the current time?", "Calculate 15 * 24", "What is 12 squared plus 8 cubed?" ] for task in tasks: print(f"\n任务: {task}") result = agent.run(task) print(f"结果: {result}") def demo_conversation_memory(): """ 演示对话内存的使用 """ print("\n=== 对话内存演示 ===") # 创建内存 memory = ConversationBufferMemory(max_history=5) # 模拟对话 conversations = [ {"question": "Hello"}, {"question": "What did I just say?"}, {"question": "Tell me about Python programming"}, {"question": "What was my last question?"}, {"question": "Summarize our conversation"} ] llm = MockLLM() qa_template = PromptTemplate( template="{history}\nHuman: {question}\nAI:", input_variables=["history", "question"] ) for i, conv in enumerate(conversations): # 加载内存 memory_vars = memory.load_memory_variables(conv) # 创建提示 prompt = qa_template.format( history=memory_vars['history'], question=conv['question'] ) # 调用LLM response = llm.predict(prompt) # 保存到内存 memory.save_context(conv, {"answer": response}) print(f"轮次 {i+1}:") print(f" 问题: {conv['question']}") print(f" 回答: {response}") print(f" 历史长度: {len(memory.history)}") print() def demo_complex_scenario(): """ 演示复杂场景：构建一个智能助手 """ print("\n=== 复杂场景演示：智能助手 ===") # 创建组件 llm = MockLLM() tools = [CalculatorTool(), TimeTool()] memory = ConversationBufferMemory(max_history=3) # 创建Agent assistant_agent = Agent(llm, tools, max_iterations=5) # 模拟用户交互 user_queries = [ "Hello, what time is it?", "Can you calculate 25 * 4 + 10?", "Thank you for your help!" ] print("智能助手对话模拟:") for i, query in enumerate(user_queries): print(f"\n[{i+1}] 用户: {query}") # 加载内存历史 memory_vars = memory.load_memory_variables({"query": query}) # 如果有历史记录，在任务中包含历史 if memory_vars['history']: task_with_history = f"Previous conversation:\n{memory_vars['history']}\nCurrent query: {query}" else: task_with_history = query # 运行Agent response = assistant_agent.run(task_with_history) print(f" 助手: {response}") # 保存到内存 memory.save_context({"query": query}, {"response": response}) # 显示内存状态 print(f" 内存历史长度: {len(memory.history)}") class DocumentLoader: """ 文档加载器模拟 """ def __init__(self): self.documents = { "python_basics.txt": "Python is a high-level programming language known for its simplicity and readability...", "ai_ethics.txt": "Artificial intelligence ethics is a branch of ethics that studies the moral implications of AI technologies...", "web_dev.txt": "Web development involves creating websites and web applications using various technologies..." } def load_document(self, doc_id: str) -> Optional[str]: """ 加载文档 """ return self.documents.get(doc_id) def search_documents(self, query: str) -> List[Dict[str, Any]]: """ 搜索文档 """ results = [] query_lower = query.lower() for doc_id, content in self.documents.items(): if query_lower in content.lower(): results.append({ 'document_id': doc_id, 'content': content, 'similarity': len(set(query_lower.split()) & set(content.lower().split())) / len(set(query_lower.split())) }) # 按相似度排序 results.sort(key=lambda x: x['similarity'], reverse=True) return results def demo_document_qa(): """ 演示文档问答 """ print("\n=== 文档问答演示 ===") # 创建文档加载器 doc_loader = DocumentLoader() # 创建LLM和提示模板 llm = MockLLM() qa_template = PromptTemplate( template="Based on the following document:\n{document}\n\nAnswer this question: {question}", input_variables=["document", "question"] ) # 问答示例 questions = [ "What is Python?", "What is AI ethics?", "What is web development?" ] for question in questions: print(f"\n问题: {question}") # 搜索相关文档 search_results = doc_loader.search_documents(question) if search_results: # 使用最相关的文档 best_doc = search_results[0] print(f"使用文档: {best_doc['document_id']}") # 创建提示 prompt = qa_template.format( document=best_doc['content'][:200] + "...", # 截断显示 question=question ) # 获取答案 answer = llm.predict(prompt) print(f"答案: {answer}") else: print("未找到相关文档") def demo_chain_of_thought(): """ 演示思维链 """ print("\n=== 思维链演示 ===") llm = MockLLM() # 思维链提示模板 cot_template = PromptTemplate( template="""Let's solve this step by step: Question: {question} Step 1: Understand what is being asked <thinking> {question} </thinking> Step 2: Break down the problem <thinking> Identify key components of the problem </thinking> Step 3: Solve the problem <thinking> Work through the solution </thinking> Final Answer: """, input_variables=["question"] ) questions = [ "If a train travels 60 mph for 2 hours, how far does it travel?", "What day will it be 100 days from today?" ] for question in questions: print(f"\n问题: {question}") prompt = cot_template.format(question=question) response = llm.predict(prompt) print(f"思维链响应: {response}") if __name__ == "__main__": demo_prompt_templates() demo_simple_chain() demo_tools() demo_agent() demo_conversation_memory() demo_complex_scenario() demo_document_qa() demo_chain_of_thought() print("\n=== LangChain框架演示完成 ===") print("本演示展示了LangChain的核心概念和组件:") print("1. Prompt Templates - 提示模板管理") print("2. LLMs - 语言模型接口") print("3. Tools - 可调用工具") print("4. Chains - 组件链接") print("5. Agents - 智能代理") print("6. Memory - 对话记忆")

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】