使用LangChain创建智能Agent完整指南

使用LangChain创建智能Agent完整指南

本文将详细介绍如何使用LangChain框架创建一个功能完整的智能Agent，该Agent能够调用MCP服务、管理对话上下文并提供智能问答功能。

项目概述

我们创建了一个基于LangChain的智能Agent系统，具有以下特性：

1. 支持调用MCP服务（时间服务和天气服务）
2. 实现混合内存管理（同时使用ConversationBufferWindowMemory和ConversationSummaryMemory）
3. 提供交互式聊天模式
4. 支持多种内存类型（buffer_window、summary、hybrid）

核心组件

1. MCP服务

我们实现了两个MCP服务：

• 时间服务：获取指定时区的当前时间
• 天气服务：通过高德API获取指定城市的天气信息

2. Agent核心类

class MCPServiceAgent:def __init__(self, memory_type: str = "hybrid", k: int = 5, max_token_limit: int = 2000):# 初始化LLMself.llm = VolcengineLLM(streaming=False)# 初始化内存self.memory_type = memory_typeif memory_type == "summary":self.summary_memory = ConversationSummaryMemory(...)self.buffer_memory = Noneelif memory_type == "buffer_window":self.buffer_memory = ConversationBufferWindowMemory(...)self.summary_memory = Noneelse:  # 默认使用hybrid（混合模式）self.buffer_memory = ConversationBufferWindowMemory(...)self.summary_memory = ConversationSummaryMemory(...)

3. 混合内存管理

我们实现了三种内存管理模式：

1. Buffer Window Memory：保存最近K轮完整对话
2. Summary Memory：使用LLM生成对话摘要
3. Hybrid Memory：同时使用以上两种内存

在混合模式下，Agent可以：

• 保留最近5轮完整对话用于精确上下文理解
• 生成历史对话摘要用于长期记忆管理

实现细节

Prompt模板设计

MCP_AGENT_PROMPT = PromptTemplate.from_template("""你是一个智能Agent，可以调用各种MCP服务来帮助用户解决问题。当前时间: {current_time}对话历史:
{history}可用的工具:
{tools}工具使用指南:
1. get_current_time: 获取指定时区的当前时间
2. get_weather: 获取指定城市的天气信息使用以下格式:
Question: 用户的问题
Thought: 你应该总是思考下一步该做什么
Action: 要执行的操作，应该是[{tool_names}]中的一个
Action Input: 操作的输入
Observation: 操作的结果
... (这个Thought/Action/Action Input/Observation可以重复N次)
Thought: 我现在知道最终答案了
Final Answer: 对原始输入问题的最终答案开始!Question: {input}
Thought: {agent_scratchpad}"""
)

工具注册

def _register_mcp_tools(self) -> List[Tool]:def get_current_time_func(input_str: str) -> str:# 实现时间获取逻辑passdef get_weather_func(input_str: str) -> str:# 实现天气获取逻辑passtools = [Tool(name="get_current_time",func=get_current_time_func,description="获取指定时区的当前时间"),Tool(name="get_weather",func=get_weather_func,description="获取指定城市的天气信息")]return tools

内存管理

在混合模式下，我们同时管理两种内存：

def _get_history(self) -> str:if self.memory_type == "hybrid":# 混合模式：结合buffer和summarybuffer_vars = self.buffer_memory.load_memory_variables({}) if self.buffer_memory else {}summary_vars = self.summary_memory.load_memory_variables({}) if self.summary_memory else {}buffer_history = buffer_vars.get("history", "")summary_history = summary_vars.get("summary", "")# 组合历史记录if summary_history and buffer_history:return f"之前的对话摘要:\n{summary_history}\n\n最近的详细对话:\n{buffer_history}"elif buffer_history:return buffer_historyelif summary_history:return summary_historyelse:return "无对话历史"

VolcengineLLM实现

我们实现了火山引擎大语言模型的LangChain集成，支持流式和非流式输出：

class VolcengineLLM(LLM):"""火山引擎大语言模型的LangChain集成"""# 火山API的访问密钥api_key: str = "YOUR_API_KEY_HERE"# 火山API的端点URLendpoint: str = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"# 模型名称model_name: str = "doubao-1-5-lite-32k-250115"# 温度参数，控制生成的随机性temperature: float = 0.7# 最大生成token数max_tokens: int = 1024# 是否启用流式输出streaming: bool = False@propertydef _llm_type(self) -> str:return "doubao-1-5-lite-32k-250115"def _construct_payload(self, prompt: str, stream: bool = False) -> Dict:"""构建API请求参数"""return {"model": self.model_name,"messages": [{"role": "user", "content": prompt}],"temperature": self.temperature,"max_tokens": self.max_tokens,"stream": stream}def _call(self, prompt: str, stop: Optional[List[str]] = None, run_manager=None, **kwargs) -> str:"""调用火山API并返回结果"""if self.streaming:# 如果启用流式输出，则使用流式处理combined_text = ""for chunk in self._stream(prompt, stop, run_manager=run_manager, **kwargs):combined_text += chunk.textreturn combined_textelse:# 否则使用常规的非流式处理return self._non_stream_call(prompt, stop)def _non_stream_call(self, prompt: str, stop: Optional[List[str]] = None) -> str:"""非流式调用火山API并返回结果"""headers = {"Content-Type": "application/json","Authorization": f"Bearer {self.api_key}"}payload = self._construct_payload(prompt, False)try:response = requests.post(url=self.endpoint,headers=headers,data=json.dumps(payload),timeout=60,verify=True)response.raise_for_status()result = response.json()# 根据火山API的实际响应格式解析结果if "choices" in result and len(result["choices"]) > 0:return result["choices"][0]["message"]["content"]else:return f"API响应格式异常: {json.dumps(result, ensure_ascii=False)}"except requests.exceptions.RequestException as e:return f"API调用失败: {str(e)}"except Exception as e:return f"未知错误: {str(e)}"def _stream(self, prompt: str, stop: Optional[List[str]] = None, run_manager=None, **kwargs) -> Iterator[GenerationChunk]:"""流式调用火山API并返回结果"""headers = {"Content-Type": "application/json","Authorization": f"Bearer {self.api_key}","Accept": "text/event-stream"}payload = self._construct_payload(prompt, True)try:response = requests.post(url=self.endpoint,headers=headers,data=json.dumps(payload),timeout=60,verify=True,stream=True)response.raise_for_status()# 处理流式响应for line in response.iter_lines():if line:decoded_line = line.decode('utf-8')if decoded_line.startswith('data:'):data_str = decoded_line[5:].strip()if data_str == '[DONE]':breaktry:data = json.loads(data_str)if "choices" in data and len(data["choices"]) > 0:delta = data["choices"][0].get("delta", {})content = delta.get("content", "")if content:yield GenerationChunk(text=content)except json.JSONDecodeError:continueexcept requests.exceptions.RequestException as e:error_msg = f"API调用失败: {str(e)}"yield GenerationChunk(text=error_msg)except Exception as e:error_msg = f"未知错误: {str(e)}"yield GenerationChunk(text=error_msg)@propertydef _identifying_params(self) -> Mapping[str, Any]:"""返回用于识别LLM的参数"""return {"model_name": self.model_name,"temperature": self.temperature,"max_tokens": self.max_tokens,"streaming": self.streaming}

使用方法

启动Agent

# 创建MCP服务Agent，使用混合内存模式（默认）
agent = MCPServiceAgent(memory_type="hybrid", k=5)# 启动交互式聊天模式
agent.chat()

交互命令

在聊天模式下，支持以下命令：

• quit/exit/退出：结束对话
• help/帮助：查看可用工具
• memory：查看当前内存中的上下文
• clear：清除对话历史

总结

通过本文介绍的方法，我们成功创建了一个功能完整的LangChain Agent，具有以下优势：

1. 灵活的内存管理：支持多种内存模式，满足不同场景需求
2. 强大的工具调用能力：可以轻松扩展新的MCP服务
3. 良好的交互体验：提供友好的命令行交互界面
4. 健壮的错误处理：具备完善的异常处理机制

这个Agent框架可以作为构建更复杂AI应用的基础，通过添加更多MCP服务和优化Prompt设计，可以进一步提升Agent的智能水平。