从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( model=self.config.llm.model_name, # 模型名称 base_url=self.config.llm.base_url, # API 地址 api_key=self.config.llm.api_key, # 密钥 temperature=self.config.llm.temperature, # 创造性程度 max_tokens=self.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}, timeout=60.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_name="chat_history"), # 对话历史占位 ("human", "{input}"), # 用户输入 MessagesPlaceholder(variable_name="agent_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)}", }