从0到1构建奶牛行为智能监控系统（七）基于langchain的智能体构建

前面已经在服务器实现了基于vllm的qwen大模型部署目标检测、身份识别、行为识别服务部署数据库服务也搭建起来了这期就介绍使用langchain构建智能体将服务封装为工具供大模型调用一、langchain介绍与安装langchain智能体开发框架的内部封装好了很多智能体开发工具可以有效降低开发复杂度。本项目使用了langchain v0.3版本安装方式参考官网LangChain 中文教程 | LangChain 中文文档二、智能体构建2.1.大模型初始化配置智能体所使用的大模型这里就是我们前面部署的qwen3.5-9b模型使用vllm部署的适配openai调用方式def _create_llm(self) - ChatOpenAI: return ChatOpenAI( modelself.config.llm.model_name, # 模型名称 base_urlself.config.llm.base_url, # API 地址 api_keyself.config.llm.api_key, # 密钥 temperatureself.config.llm.temperature, # 创造性程度 max_tokensself.config.llm.max_tokens, # 最大输出长度 extra_body{ chat_template_kwargs: { enable_thinking: self.config.llm.enable_thinking, } }, )2.2.定义目标检测、身份识别、行为识别服务工具集这里将之前部署的服务包装为智能体能够调用的工具def _create_tools(self) - list: tools [] tools.extend(get_detection_tools()) # 目标检测工具 tools.extend(get_behavior_tools()) # 行为分析工具 tools.extend(get_identification_tools()) # 身份识别工具 tools.extend(get_database_tools()) # 数据库查询工具 return tools以身份识别为例通过添加tool修饰器langchain会自动将方法注册为工具方法内部使用httpx库进行http请求发送调用我们前面部署好的各种模型服务from langchain_core.tools import tool import httpx tool def identify_cow(topk: int 5) - str: 根据最近上传的奶牛图像进行个体身份识别。 url f{IdentificationToolConfig.base_url}/api/identifi/detect try: # 调用识别服务 response httpx.post(url, params{topk: topk}, timeout60.0) response.raise_for_status() data response.json() if not data.get(success): return f识别失败{data.get(message, 未知错误)} # 格式化结果 results data.get(results, []) if not results: return 未找到匹配的奶牛。 lines [奶牛身份识别结果] for i, r in enumerate(results, 1): lines.append( f\n匹配 {i}ID{r.get(id)}, f名称{r.get(name)}, f置信度{r.get(confidence)}% ) return \n.join(lines) except httpx.ConnectError: return 身份识别服务未启动。 except httpx.TimeoutException: return 请求超时请稍后重试。 except Exception as exc: return f识别出错{exc}2.3.提示词构建与智能体构建这里自行写入提示词然后使用langchain提供的方法将大模型、工具和提示词连接起来构建成智能体prompt ChatPromptTemplate.from_messages([ (system, self.config.system_prompt), # 系统角色设定 MessagesPlaceholder(variable_namechat_history), # 对话历史占位 (human, {input}), # 用户输入 MessagesPlaceholder(variable_nameagent_scratchpad), # 工具调用痕迹 ]) //智能体构建 agent create_tool_calling_agent(self.llm, self.tools, prompt)2.5.主要对话接口对话调用输入前端传入的用户信息返回智能体结果def chat( self, message: str, chat_history: list[BaseMessage] | None None, ) - dict[str, Any]: 主聊天接口 if chat_history is None: chat_history [] try: result self.agent_executor.invoke({ input: message, chat_history: chat_history, }) return { success: True, response: result.get(output, ), } except Exception as e: return { success: False, error: str(e), response: f处理请求时发生错误{str(e)}, }