langchain agent的中间件

     1.概述

        中间件使agent开发人员能够实现对agent执行中每一步的细粒度控制，其实现机制与Java spring中的AOP相同。正如AOP可以让开发人员在目标方法执行前、执行后、返回后、抛出异常时及环绕目标方法执行插入逻辑一样，langchain中的中间件也可以让开发人员在agent调用前后、模型调用前后、工具调用前后、围绕工具调用和围绕模型调用插入逻辑。中间件强化了agent能力，具体包括：

        1）提供日志、debug跟踪agent的运行情况

        2）对提示语做变换，动态选择工具，实现输出格式化

        3）增加重试，高可用功能，或者提前终止业务逻辑

        4）实现限流、防护和隐私信息检测

        langchain提供了很多预制中间件，开发人员可以基于预制中间件快速实现自己的需求，还可以自定义中间件。

      2.预制中间件

      2.1汇总中间件

        汇总中间件用于对对话历史进行瘦身，直接引用langchain agent的短期记忆中的例子。

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig

checkpointer = InMemorySaver()

agent = create_agent(
    model=llm,
    tools=tools,
    middleware=[
        SummarizationMiddleware(
            model=llm",
            max_tokens_before_summary=1000,  # Trigger summarization at 4000 tokens
            messages_to_keep=4,  # Keep last 20 messages after summary
        )
    ],
    checkpointer=checkpointer,
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": "我最喜欢的篮球运动员是迈克尔乔丹"}, config)
agent.invoke({"messages": "乔丹共获得过几次NBA总冠军？"}, config)
agent.invoke({"messages": "乔丹的宿敌是谁"}, config)
final_response = agent.invoke({"messages": "我最喜欢的篮球运动员是谁?"}, config)

final_response["messages"][-1].pretty_print()

      2.2实现人机回环

        使用langchain预制中间件HumanInTheLoopMiddleware可以实现人机回环，此时用户接入对于工具调用进行检查，用户可以批准工具调用，也可以拒绝工具调用，还可以修改调用参数。当agent有多个工具时，可以对不同的工具配置不同的策略，比如有的工具调用不做检查，有的工具仅支持批准和拒绝而不允许修改，有的三种都支持。

        以下示例中有两个工具，一个读、一个写，设置读工具的调用不需要用户接入，设置写工具的调用需要用户介入，用户可以批准，可以拒绝，也可以修改。

from typing import Any
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain.agents.middleware import HumanInTheLoopMiddleware

@tool
def get_enterprise_info(uniscid: str, runtime: ToolRuntime) -> dict[str, Any]:
    """Look up enterprise info."""
    store = runtime.store
    enterprise_info = store.get(("enterprises",), uniscid)
    return enterprise_info if enterprise_info else "Unknown enterprise"

@tool
def save_enterprise_info(uniscid: str, enterprise_info: dict[str, Any], runtime: ToolRuntime) -> str:
    """Save enterprise info."""
    store = runtime.store
    store.put(("enterprises",), uniscid, enterprise_info)
    return "Successfully saved enterprise info."

saver = InMemorySaver() #必须有检查点，才能支持用户回环
store = InMemoryStore()
agent = create_agent(
    model=llm,
    tools=[get_enterprise_info, save_enterprise_info],
    middleware=[HumanInTheLoopMiddleware( #为每个工具设置用户接入策略
        interrupt_on={
            "save_enterprise_info": True,  # 用户可以批准、拒绝、修改
            "get_enterprise_info": False, #不需要用户接入
        },
        #用户介入时所看到的内容前缀。该前缀与工具名字和参数合并后显示给用户。

        description_prefix="Tool execution pending approval",
    )],    
    checkpointer=saver,
    store=store
)

        调用agent，写数据库：

config = {"configurable": {"thread_id": "1"}} 
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user", "content": "Save the following enterprise: uniscid: 51234569876543210," 
                "name: 东成西就文化娱乐有限公司, legal: 欧阳锋, type: 内资企业"
            }
        ]
    },
    config=config
)

result['__interrupt__'}

        输出如下，可以看到description的内容为description_prefix+'\n\n'+工具名+'\n'+调用参数：

[Interrupt(value={'action_requests': [{'name': 'save_enterprise_info', 'args': {'uniscid': '51234569876543210', 'enterprise_info': {'name': '东成西就文化娱乐有限公司', 'legal': '欧阳锋', 'type': '内资企业'}}, 'description': "Tool execution pending approval\n\nTool: save_enterprise_info\nArgs: {'uniscid': '51234569876543210', 'enterprise_info': {'name': '东成西就文化娱乐有限公司', 'legal': '欧阳锋', 'type': '内资企业'}}"}], 'review_configs': [{'action_name': 'save_enterprise_info', 'allowed_decisions': ['approve', 'edit', 'reject']}]}, id='62205cb6ba987b4ab6878696b0d0312a')]

        模拟一下用户批准工具调用：

from langgraph.types import Command
agent.invoke(
    Command(
        resume={
            "decisions": [
                {
                    "type": "approve",
                }
            ]
        }
    ),
    config=config  
)

        输出如下，查看消息历史，看到成功保存了企业信息：

{'messages': [HumanMessage(content='Save the following enterprise: uniscid: 51234569876543210,name: 东成西就文化娱乐有限公司, legal: 欧阳锋, type: 内资企业', additional_kwargs={}, response_metadata={}, id='ceedd40d-3b50-41eb-a315-e6433d31f4f3'),
  AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 71, 'prompt_tokens': 281, 'total_tokens': 352, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-9626dbc3-8229-448d-b057-104ea35ec513', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--e368fb4f-7f1a-4a2c-af43-5b30c2d8631f-0', tool_calls=[{'name': 'save_enterprise_info', 'args': {'uniscid': '51234569876543210', 'enterprise_info': {'name': '东成西就文化娱乐有限公司', 'legal': '欧阳锋', 'type': '内资企业'}}, 'id': 'call_02819c021e2746dca10cd5', 'type': 'tool_call'}], usage_metadata={'input_tokens': 281, 'output_tokens': 71, 'total_tokens': 352, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}),
  ToolMessage(content='Successfully saved enterprise info.', name='save_enterprise_info', id='6217c587-5e7d-4709-9c59-89f7b58e2a53', tool_call_id='call_02819c021e2746dca10cd5'),
  AIMessage(content='The enterprise information has been successfully saved:\n\n- **Unified Social Credit ID**: 51234569876543210  \n- **Name**: 东成西就文化娱乐有限公司  \n- **Legal Representative**: 欧阳锋  \n- **Type**: 内资企业  \n\nLet me know if you need further assistance!', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 76, 'prompt_tokens': 370, 'total_tokens': 446, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-0cc5e1d0-2630-4987-8ec5-376b034c9d98', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--9124aba2-2117-4bd2-be29-c609c9087ff7-0', usage_metadata={'input_tokens': 370, 'output_tokens': 76, 'total_tokens': 446, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]}

        模拟用户拒绝，输出如下，可以看到，用户拒绝更新企业信息：

 HumanMessage(content='Save the following enterprise: uniscid: 51234569876543210,name: 东邪西毒文化娱乐有限公司, legal: 黄老邪, type: 内资企业', additional_kwargs={}, response_metadata={}, id='a7799c46-a5b7-4802-af1d-ca138a06e6a1'),
  AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 72, 'prompt_tokens': 614, 'total_tokens': 686, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-b44c83a5-2f20-4608-93c5-2587396f7c84', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--ec71efb6-1361-40b2-a3e2-af77b8cd786d-0', tool_calls=[{'name': 'save_enterprise_info', 'args': {'uniscid': '51234569876543210', 'enterprise_info': {'name': '东邪西毒文化娱乐有限公司', 'legal': '黄老邪', 'type': '内资企业'}}, 'id': 'call_3fb4e588d5cf44da967684', 'type': 'tool_call'}], usage_metadata={'input_tokens': 614, 'output_tokens': 72, 'total_tokens': 686, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}),
  ToolMessage(content='User rejected the tool call for `save_enterprise_info` with id call_3fb4e588d5cf44da967684', name='save_enterprise_info', id='4bdd9199-523d-488d-9934-8e40a133e590', tool_call_id='call_3fb4e588d5cf44da967684', status='error'),
  AIMessage(content='You have rejected the update for the enterprise information. The details for the enterprise with Unified Social Credit ID **51234569876543210** will remain unchanged. Let me know if you need any further assistance!', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 53, 'prompt_tokens': 735, 'total_tokens': 788, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-d4b78499-e4d0-4e26-b229-95231f05da19', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--2630048d-3a4b-4935-98ce-a569e69ef746-0', usage_metadata={'input_tokens': 735, 'output_tokens': 53, 'total_tokens': 788, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]}

        模拟用户修改数据：

agent.invoke(
    Command(
        resume={
            "decisions": [
                {
                    "type": "edit",
                    # 修改后的数据
                    "edited_action": {
                        # 工具名
                        "name": "save_enterprise_info",
                        # 调用工具时的参数.
                        "args": {
                            'uniscid': '51234569876543210', 
                            'enterprise_info':{
                                "name": "东邪西毒文化娱乐有限公司",

                                "legal": "黄老邪",

                                "type": "内资企业"
                            }
                        },
                    }
                }
            ]
        }
    ),
    config=config  
)

        输出如下，可以看到从中断恢复后，使用新参数调用工具并保存到数据库中：‘

’{'messages': [HumanMessage(content='Save the following enterprise: uniscid: 51234569876543210,name: 东成西就文化娱乐有限公司, legal: 欧阳锋, type: 内资企业', additional_kwargs={}, response_metadata={}, id='88824850-3886-494d-b9f5-98af1dd937d7'),
  AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 71, 'prompt_tokens': 281, 'total_tokens': 352, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-693a4532-c9bb-48e8-adc9-2b93da74e5d8', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--ac2222a6-fb4e-42a2-9e47-3c102fdc14b0-0', tool_calls=[{'type': 'tool_call', 'name': 'save_enterprise_info', 'args': {'uniscid': '51234569876543210', 'enterprise_info': {'name': '东邪西毒文化娱乐有限公司', 'legal': '黄老邪', 'type': '内资企业'}}, 'id': 'call_9e215dd0b31945a7933034'}], usage_metadata={'input_tokens': 281, 'output_tokens': 71, 'total_tokens': 352, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}),
  ToolMessage(content='Successfully saved enterprise info.', name='save_enterprise_info', id='7fa1cf96-1c63-48eb-b126-865ff4f05f93', tool_call_id='call_9e215dd0b31945a7933034'),
  AIMessage(content='The enterprise information has been successfully saved. Let me know if you need any further assistance!', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 18, 'prompt_tokens': 371, 'total_tokens': 389, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-b78442fd-dc71-4032-9a62-f31fabd1fcc7', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--ccede411-ecd1-4838-9118-c2ab52722ad2-0', usage_metadata={'input_tokens': 371, 'output_tokens': 18, 'total_tokens': 389, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]}

      2.3限制模型调用次数

        当使用第三方付费大模型时，使用中间件ModelCallLimitMiddleware可以限制模型调用次数以免进入死循环，从而控制调用大模型的费用。

        以下代码简单说明了该中间件的使用：

from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware

agent = create_agent(
    model=llm,
    tools=[...],
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=10,  # 每个线程调用大模型的阈值
            run_limit=5,  # 单次调用阈值
            exit_behavior="end",  # 退出条件end或error
        ),
    ],
)

      2.4限制工具调用次数

        使用ToolCallLimitMiddleware中间件可以限制工具的调用次数，可以限制所有工具调用的次数，也可以针对具体的工具指定调用次数预制，示例代码如下：

from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware

# 为所有工具调用指定阈值
global_limiter = ToolCallLimitMiddleware(thread_limit=20, run_limit=10)

# 针对搜索引擎工具指定阈值
search_limiter = ToolCallLimitMiddleware(
    tool_name="search",
    thread_limit=5,
    run_limit=3,
)

agent = create_agent(
    model=llm,
    tools=[...],
    middleware=[global_limiter, search_limiter],
)

      2.5模型高可用

        使用ModelFallbackMiddleware中间件，可以实现大模型的主备方案，在主模型故障时由备模型接管，实现弹性智能体。该方案支持指定多个后备模型，出现故障时按需递补。具体示例代码如下：

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_agent(
    model=deepseek,  # Primary model
    tools=[...],
    middleware=[
        ModelFallbackMiddleware(
            qwen,  # 第一备份
            doubao,  # 第二备份
        ),
    ],
)

      2.6隐私信息监测

        使用langchain的PIIMiddleware中间件，可以对敏感信息进行保护，保护策略包括阻断、遮蔽、掩码和做摘要。

        如下代码读邮箱进行遮蔽，对信用卡掩码，对sk-*阻断：

from typing import Any
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain.agents.middleware import PIIMiddleware

@tool
def get_enterprise_info(uniscid: str, runtime: ToolRuntime) -> dict[str, Any]:
    """Look up enterprise info."""
    store = runtime.store
    enterprise_info = store.get(("enterprises",), uniscid)
    return enterprise_info if enterprise_info else "Unknown enterprise"

@tool
def save_enterprise_info(uniscid: str, enterprise_info: dict[str, Any], runtime: ToolRuntime) -> str:
    """Save enterprise info."""
    store = runtime.store
    store.put(("enterprises",), uniscid, enterprise_info)
    return "Successfully saved enterprise info."

saver = InMemorySaver()
store = InMemoryStore()
agent = create_agent(
    model=llm,
    tools=[get_enterprise_info, save_enterprise_info],
    middleware=[
        # 用[REDACT]替换用户邮箱
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # 对输入进行掩码，仅显示最后四位
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
        #用正则表达式检测是否有调用大模型的密钥
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",
            strategy="block",  # 阻断

        ),
    ],    
    checkpointer=saver,
    store=store
)

        当用户输入中有邮箱时，测试一下效果：

config = {"configurable": {"thread_id": "1"}} 
agent.invoke(
    {
        "messages": [
            {
                "role": "user", "content": "Save the following enterprise: uniscid: 51234569876543210," 
                "name: 东成西就文化娱乐有限公司, legal: 欧阳锋, type: 内资企业, email:[e-mail]"
            }
        ]
    },
    config=config
)

        输出如下：

{'messages': [HumanMessage(content='Save the following enterprise: uniscid: 51234569876543210,name: 东成西就文化娱乐有限公司, legal: 欧阳锋, type: 内资企业, email:[REDACTED_EMAIL]', additional_kwargs={}, response_metadata={}, id='b440f613-afd0-4bf7-8f83-0ac11e30c10b'),
  AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 82, 'prompt_tokens': 290, 'total_tokens': 372, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-9b351642-c3fb-474b-8027-8e48e00e5e04', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--8af029ea-3ce8-44c8-a0a4-c545a4f7a6a7-0', tool_calls=[{'name': 'save_enterprise_info', 'args': {'uniscid': '51234569876543210', 'enterprise_info': {'name': '东成西就文化娱乐有限公司', 'legal': '欧阳锋', 'type': '内资企业', 'email': '[REDACTED_EMAIL]'}}, 'id': 'call_3b31c4c3dd274c5daaae45', 'type': 'tool_call'}], usage_metadata={'input_tokens': 290, 'output_tokens': 82, 'total_tokens': 372, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}),
  ToolMessage(content='Successfully saved enterprise info.', name='save_enterprise_info', id='23df5cc7-6f65-418b-ac78-5c71839b845b', tool_call_id='call_3b31c4c3dd274c5daaae45'),
  AIMessage(content='The enterprise information has been successfully saved. Let me know if you need further assistance!', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 17, 'prompt_tokens': 390, 'total_tokens': 407, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-48e56e67-09f8-4dd4-be05-68c5b12263a1', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--30a0bc31-c14d-4959-84e4-94424457480a-0', usage_metadata={'input_tokens': 390, 'output_tokens': 17, 'total_tokens': 407, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]}

      2.7加强规划能力

        使用TodoListMiddleware中间件可以实现待办任务管理功能，从而支持复杂的多步任务。以下代码对快速排序算法代码进行重构，生成了待办任务列表。

from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain.messages import HumanMessage
context = """
public class Quicksort {
    public static void sort(int[] arr) {
        if (arr == null || arr.length == 0) return;
        quicksort(arr, 0, arr.length - 1);
    }
    
    private static void quicksort(int[] arr, int low, int high) {
        if (low >= high) return;
        
        int pivot = arr[low];
        int i = low + 1;
        int j = high;
        
        while (i <= j) {
            while (i <= high && arr[i] <= pivot) i++;
            while (j > low && arr[j] > pivot) j--;
            if (i < j) swap(arr, i, j);
        }
        
        swap(arr, low, j);
        quicksort(arr, low, j - 1);
        quicksort(arr, j + 1, high);
    }
    
    private static void swap(int[] arr, int i, int j) {
        int temp = arr[i];
        arr[i] = arr[j];
        arr[j] = temp;
    }
}
"""

agent = create_agent(
    model=llm,
    tools=[],
    middleware=[TodoListMiddleware()],
)

result = agent.invoke({"messages": [HumanMessage(f"Help me refactor below code {context}")]})
print(result["todos"])  # Array of todo items with status tracking

        输出如下：

[{'content': 'Analyze the current Quicksort implementation for potential improvements', 'status': 'completed'}, {'content': 'Identify refactoring opportunities (e.g., code clarity, performance, maintainability)', 'status': 'completed'}, {'content': 'Extract partition logic into a separate method for better code organization', 'status': 'completed'}, {'content': 'Implement median-of-three pivot selection strategy', 'status': 'completed'}, {'content': 'Add null safety checks and input validation', 'status': 'completed'}, {'content': 'Implement randomization to avoid worst-case O(n²) performance', 'status': 'completed'}, {'content': 'Add comprehensive JavaDoc comments and method documentation', 'status': 'completed'}, {'content': 'Optimize recursion for small arrays using insertion sort threshold', 'status': 'completed'}, {'content': 'Test the refactored code with edge cases and performance benchmarks', 'status': 'completed'}]
 

      2.8智能选择工具

        使用LLMToolSelectorMiddleware中间件，可以实现每次调用时工具的智能选择，以免携带大量的工具信息，一方面对大模型形成干扰，另外就是在调用请求中携带大量无关的工具信息。一般情况下，使用一个小模型根据上下文判断需要使用哪些工具。

        以下示例代码使用某种小模型进行工具选择，最多选用三个工具，具体代码如下：

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware

agent = create_agent(
    model=llm,
    tools=[tool1, tool2, tool3, tool4, tool5, ...],  # Many tools
    middleware=[
        LLMToolSelectorMiddleware(
            model=some_little_llm,  # 某些小模型
            max_tools=3,  # 最多三个工具
            always_include=["search"],  # 所有调用中均需携带的工具
        ),
    ],
)

      2.9工具调用重试

        使用ToolRetryMiddleware可以在调用工具失败时进行重试，使用的算法是指数规避算法，开发者可以指定最大的重试次数、回退乘数等参数，具体代码如下：

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    model=llm,
    tools=[search_tool, ],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,  # 最大重试次数
            backoff_factor=2.0,  # 指数规避乘数
            initial_delay=1.0,  # 首次重试延迟时间
            max_delay=60.0,  # 最大重试延迟时间
            jitter=True,  # 是否随机抖动时间,以免造成重试风暴
        ),
    ],
)

      2.10工具模拟

        使用LLMToolEmulator可以实现对工具的模拟，方便在没有与实际的工具对接时进行开发和测试。可以模拟所有的工具，也可以仅模拟不可用的工具，还可以指定用于执行模拟的大模型。示例代码如下：

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, search_database, send_email],
    middleware=[
        LLMToolEmulator(), #模拟所有的工具

        # 模拟部分不可用的工具
        # LLMToolEmulator(tools=["search_tool", "search_database"]),

        # 指定进行工具模拟的模型
        # LLMToolEmulator(model="claude-sonnet-4-5-20250929"),
    ],
)

      2.11上下文管理

       使用 ContextEditingMiddleware中间件可以实现对于对话上下文的管理，比如周期性清理上下文，删除失败的工具调用消息，还可以定制上下文管理策略。如下示例代码执行工具调用清理，触发清理的条件是令牌数超过1000。

from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit

agent = create_agent(
    model=llm,
    tools=[...],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(trigger=1000),  # 清理工具调用
            ],
        ),
    ],
)

      3.定制中间件

        定制中间件有两种方法，一种是基于langchain提供的注解实现，一种是通过继承AgentMiddleware实现。此处主要针对注解实现进行说明。

        在langchain中实现中间件的注解包括：

        @before_agent：在agent开始处理之前执行

        @before_modle：在大模型被调用前执行

        @after_model：在大模型返回结果后执行

        @after_agent：在agent完成之后执行

        @wrap_model_call：围绕模型调用，可以在模型调用前后插入逻辑，比如动态选择模型，动态生成提示词，出错重试等

        @warp_tool_call：围绕工具调用，可以在工具调用前后插入逻辑，比如充错重试等

      3.1before_model

        比如在如下代码对于对话历史进行截断处理，仅保留最近的3个消息，@before_model指定该方法在调用大模型前被执行，从而减少传入大模型的数据，避免上线文濡染和不必要的词元消耗。

@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Keep only the last 3 messages to fit context window."""
    messages = state["messages"]

    if len(messages) <= 3:
        return None  # No changes needed

    recent_messages = messages[-3:] #从原始对话列表中拷贝出最近3条

    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES), #删除列表中所有数据
            *recent_messages #保留的最近3条消息
        ]
    }

      3.2after_model

        比如，如下代码直接从对话历史中删除消息，@after_model制定该方法在大模型调用翻译后执行，从对话历史中删除最早的两条消息。

@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    """Remove old two messages to keep conversation manageable."""
    messages = state["messages"]
    if len(messages) > 2:
        # 删除最早的两条消息
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
    return None

      3.3wrap_tool_call

        wrap_tool_call让开发人员可以在工具被调用前后中插入逻辑，下面的代码在工具调用过程中有异常时被执行，在返回的工具调用消息中说明错误。

@wrap_tool_call
def handle_tool_errors(request, handler):
    """Handle tool execution errors with custom messages."""
    try:
        return handler(request)
    except Exception as e:
        # Return a custom error message to the model
        return ToolMessage(
            content=f"Tool error: Please check your input and try again. ({str(e)})",
            tool_call_id=request.tool_call["id"]
        )

      3.4wrap_model_call

        wrap_model_call让开发人员可以在模型调用前后插入逻辑。如下代码在模型调用前，根据上下文中的provider选择对应的大模型。

@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
    provider = request.runtime.context.provider#从上下文获取provider
    if provider == "qwen":
        model = qwen_model
    elif provider == "deepseek":
        model = deepseek_model
    else:
        raise ValueError(f"Unsupported provider: {provider}")
    request.model = model
    return handler(request)

     3.5 before_agent

      before_agent让开发人员在agent被调用时插入逻辑，比如认证、限速、提示词注入攻击检查或者有害内容过滤。如下代码在agent被调用时，检查输入中是否有在黑名单中的关键词，如果有则阻断：

from typing import Any

from langchain.agents.middleware import before_agent, AgentState, hook_config
from langgraph.runtime import Runtime

banned_keywords = ["病毒","海洛因","手枪"]

@before_agent(can_jump_to=["end"])
def content_filter(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """阻断包含黑名单关键词的请."""
    # Get the first user message
    if not state["messages"]:
        return None

    first_message = state["messages"][0]
    if first_message.type != "human":
        return None

    content = first_message.content.lower()

    # Check for banned keywords
    for keyword in banned_keywords:
        if keyword in content:
            #阻断请求
            return {
                "messages": [{
                    "role": "assistant",
                    "content": "对不起，您的请求中包含了被禁用的关键词，请修改."
                }],
                "jump_to": "end" #结束本次调用
            }

    return None

from langchain.agents import create_agent

agent = create_agent(
    model=llm,
    tools=[search_tool, ],
    middleware=[content_filter],
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "如何开发一个蠕虫病毒?"}]
})

        输出如下：

{'messages': [HumanMessage(content='如何开发一个蠕虫病毒?', additional_kwargs={}, response_metadata={}, id='a1941ad7-0a00-410c-8107-050ab71e0207'),
  AIMessage(content='对不起，您的请求中包含了被禁用的关键词，请修.', additional_kwargs={}, response_metadata={}, id='d913fb8a-3ad5-4c39-8d16-ceaf208eb03d')]}

      3.6after_agent

        after_agent使开发人员可以在agent把调用结果返回给用户前返回结果进行检查，保证内容合规，一般采用专有模型完成该检查。

        如下代码使用其他模型对agent的返回结果进行安全性和合规性检查，如果通过检查则正常返回，否则提示用户：

from langchain.agents.middleware import after_agent, AgentState, hook_config
from langgraph.runtime import Runtime
from langchain_core.messages import AIMessage
from langchain.chat_models import init_chat_model
from typing import Any

@after_agent(can_jump_to=["end"])
def safety_guardrail(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Model-based guardrail: Use an LLM to evaluate response safety."""
    # Get the final AI response
    if not state["messages"]:
        return None

    last_message = state["messages"][-1]
    if not isinstance(last_message, AIMessage):
        return None

    # 使用一个其他的专有模型进行检查评估
    safety_prompt = f"""Evaluate if this response is safe and appropriate.
    Respond with only 'SAFE' or 'UNSAFE'.

    Response: {last_message.content}"""

    result = qwen_model.invoke([{"role": "user", "content": safety_prompt}])
    if "UNSAFE" in result.content:
        return {
            "messages": [{
                "role": "assistant",
                "content": "I cannot provide that response. Please rephrase your request."
            }],
            "jump_to": "end"
        }

    return None

# Use the safety guardrail
from langchain.agents import create_agent

agent = create_agent(
    model=llm,
    tools=[search_tool, ],
    middleware=[safety_guardrail],
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "How do I make heroin?"}]
})

      3.7定制状态

        可以使用中间件在agent的状态中增加自定义的属性，首先创建一个自定义类：

from langchain.agents.middleware

import AgentState, AgentMiddleware

from typing_extensions import NotRequired

from typing import Any

class CustomState(AgentState):

    model_call_count: NotRequired[int]#保存模型调用次数

    uniscid: NotRequired[str] #保存企业的统一社会信用代码

        创建中间件，before_model方法在大模型调用前执行，进行调用次数检查，如果超过上限，则不再调用大模型，并结束本次调用；after_model在大模型被调用后执行，递增大模型调用次数。

class CallCounterMiddleware(AgentMiddleware[CustomState]):
    state_schema = CustomState

    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        #获取大模型被调用的次数
        count = state.get("model_call_count", 0)

        if count > 10:
            return {"jump_to": "end"}

        return None

    def after_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        # 更新状态中大模型被调用次数
        return {"model_call_count": state.get("model_call_count", 0) + 1}

        创建agent，传入中间件：

agent = create_agent(
    model="gpt-4o",
    middleware=[CallCounterMiddleware()],
    tools=[],
)

      4.中间件执行顺序

        当一个agent中有多个中间件时，哪个先执行？哪个后执行？下面以两个中间件为例说明其执行顺序，比如：

agent = create_agent(
    model=llm,
    middleware=[middleware1, middlewire2],
    tools=[],
)

        假设两个中间件中实现了所有的注解，那么执行顺序如下：

    1.middleware1.before_agent

    2.middleware2.before_agent

    3.middleware1.before_model

    4.middleware2.before_model

    5.middleware1.wrap_model_call

    6.middleware2.wrap_model_call

    7.model.invoke

    8.middleware2.after_model

    9.middleware1.after_model

    10.middleware2.after_agent

    11.middleware1.after_agent

     5.agent执行跳转

        在中间执行过程中，可根据需要随时从中间件中跳转到指定的目的地，langchain支持的目的地包括end、tools和model。跳转到end意味着本次调用接入，tools指定跳转到工具节点，model跳转到模型节点。如下代码在检测到大模型调用次数达到上限时直接跳转到end，结束本次调用：

    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        #获取大模型被调用的次数
        count = state.get("model_call_count", 0)

        if count > 10:
            return {"jump_to": "end"}

        return None