摘要使用AgentMiddleware实现自定义中间件搭建一个带任务追踪的可直接运行Agent。本文代码全部可复制修改环境配置即可直接跑通。一、前置依赖二、项目文件清单三、完整代码3.1 环境配置 .env3.2 文件工具 local_file_operator.py3.3 中间件 task_track_middleware.py3.4 Agent入口 xiaoyi_agent.py3.5 langgraph.json 运行配置四、运行方式五、核心关键点效果图占位一、前置依赖安装依赖包langchain0.3.0 langchain-openai langgraph pydantic2.0 python-dotenv二、项目文件清单文件作用.env模型密钥、接口地址配置local_file_operator.py文件读写工具task_track_middleware.py自定义AgentMiddleware任务追踪中间件xiaoyi_agent.pyAgent主入口langgraph.jsonlanggraph运行配置三、完整代码3.1 环境配置 .env密钥自行填写不要直接复制公开密钥# model config BASE_LLMqwen3.5-flash BASE_VLMqwen3.5-flash MODEL_API_BASE_URL[https://xxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1](https://xxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1) OPENAI_API_KEY你的api_key # langsmith可选不需要可以删除,langsmith api_key获取https://smith.langchain.com/ LANGSMITH_API_KEY你的langsmith_key LANGCHAIN_PROJECTxiaoyi_agent_project LANGCHAIN_TRACING_V2true3.2 文件工具 local_file_operator.pyfrom langchain.tools import tool tool def save_local_file(file_path: str, content: str) - str: 将文本内容写入本地文件 with open(file_path, w, encodingutf-8) as f: f.write(content) return f成功写入目标文件: {file_path} tool def load_local_file(file_path: str) - str: 读取本地文件返回文件内容 with open(file_path, r, encodingutf-8) as f: content f.read() return f读取文件完成{file_path}\n文件内容\n{content} tool def scan_directory_files(target_dir: str) - str: 扫描指定目录下全部文件名称列表 import os file_items os.listdir(target_dir) return f扫描目录: {target_dir}\n目录内文件列表:\n{file_items}3.3 中间件 task_track_middleware.py核心继承AgentMiddleware扩展AgentState状态追加系统提示注册中间件内部工具from langchain.agents.middleware import AgentMiddleware, AgentState from langchain.tools import tool from langchain_core.messages import ToolMessage from pydantic import BaseModel, Field from typing import Literal, Annotated from langgraph.types import Command from langchain.tools import InjectedToolCallId class WorkItem(BaseModel): 单条待办任务模型 desc: str Field(description待执行工作任务描述) run_status: Literal[等待执行, 处理中, 已做完] Field(description任务运行状态) class TaskRuntimeState(AgentState): 扩展Agent状态存储任务清单 work_list: list[WorkItem] class TaskTraceMiddleware(AgentMiddleware): state_schema TaskRuntimeState def __init__(self): super().__init__() self.append_instruct 接收任务先拆解调用 refresh_work_list 创建任务清单。 每完成一个步骤立刻标记对应任务为「已做完」不要批量更新。 tool def refresh_work_list( work_list: list[WorkItem], tool_call_id: Annotated[str, InjectedToolCallId] ) - Command: 更新agent内部任务清单 return Command( update{ work_list: work_list, messages: [ToolMessage(f任务清单已更新{work_list}, tool_call_idtool_call_id)] } ) self.tools [refresh_work_list] async def awrap_model_call(self, request, handler): # 拦截模型请求追加提示词 request.system_prompt request.system_prompt \n\n self.append_instruct return await handler(request)3.4 Agent入口 xiaoyi_agent.py读取环境变量初始化LLM挂载自定义中间件构建Agent实例import os from dotenv import load_dotenv from langchain.agents import create_agent from local_file_operator import save_local_file, load_local_file, scan_directory_files from task_track_middleware import TaskTraceMiddleware from langchain_openai import ChatOpenAI # 加载.env配置 load_dotenv() def init_chat_llm(): llm_instance ChatOpenAI( modelos.getenv(BASE_LLM), base_urlos.getenv(MODEL_API_BASE_URL), api_keyos.getenv(OPENAI_API_KEY), temperature0.1 ) return llm_instance # 创建Agent挂载中间件 smart_agent create_agent( modelinit_chat_llm(), tools[save_local_file, load_local_file, scan_directory_files], system_prompt你是擅长处理本地文件的智能助手, middleware[TaskTraceMiddleware()] )3.5 langgraph.json 运行配置{ dependencies: [.], graphs: { agent: ./xiaoyi_agent.py:smart_agent }, env: .env }四、运行方式将上面全部文件放在同一个项目根目录修改.env填入自己的模型api_key和接口地址安装依赖使用langgraph命令运行langgraph dev启动成功后访问本地web界面即可对话测试。测试示例自由构思人物与事件产出三段独立纪事文稿分段单独存一个txt文件全部文稿完成后汇总情节撰写整体纪事概述。Agent会自动拆解任务生成待办列表执行写入完成后标记任务状态。五、核心关键点自定义状态类必须继承AgentState中间件必须赋值state_schema否则状态无法保存awrap_model_call是异步钩子可以修改request里的system_prompt实现动态追加提示中间件内部self.tools注册的工具会自动生效不需要在create_agent重复传入Command只能在tool装饰的函数内返回用于更新Agent状态InjectedToolCallId由框架自动注入业务调用无需传参。Agent调用refresh_work_list更新任务列表输出】中间图略过......【web对话界面截图展示任务清单更新输出】