大家好,我是小芽。
最近在做 Python Agent 开发时,LangChain经常将代码写成这样:
def create_tool(api_client): def query_order(order_id: str): return api_client.get_order(order_id) return query_order
或者:
def create_node(model): async def node(state): return await model.ainvoke(state[”messages”]) return node
很多刚接触 Python 的朋友看到这里会有点奇怪:
为什么函数里面还要定义函数?
更奇怪的是:
query_order = create_tool(api_client)
create_tool() 明明已经执行结束了,为什么 query_order() 后面还能继续使用里面的 api_client?
这背后就是 Python 一个非常重要的机制:
闭包(Closure)。
单独学习 Python 时,闭包可能只是一个语言知识点。
但到了 Agent、LangChain、LangGraph、MCP、Tool Calling 这些场景里,你会发现闭包非常常见,因为 Agent 开发天然存在大量这样的需求:
所以今天我们从最基础的:
什么是闭包?
开始,一步一步讲到:
闭包在企业级 Agent 开发中到底有什么实际作用。
一、先别管闭包,先理解函数也是对象
理解闭包之前,必须先理解 Python 的一个重要特性:
函数本身也是对象。
例如:
def hello(): print(”hello”)
这里的:
本质上是一个函数对象。
所以我们可以:
然后:
仍然会输出:
甚至函数还可以作为参数传递:
def execute(func): func()execute(hello)
也可以把函数作为返回值:
def create_func(): def hello(): print(”hello”) return hello
然后:
func = create_func()func()
输出:
这就是理解闭包的第一步:
Python 中的函数,可以像普通变量一样被传递和返回。
二、什么是闭包?
先看一个最简单的例子。
def outer(): name = ”小芽” def inner(): print(name) return inner
调用:
然后:
输出:
这里最值得注意的事情是:
其实早就执行完了。
正常情况下,你可能会认为:
应该已经随着 outer() 执行结束而消失了。
但是:
仍然可以访问 name。
为什么?
因为 inner() 不只是把自己的函数代码保存了下来。
它还把自己依赖的外部变量:
一起保存了下来。
这就是闭包。
可以简单理解为:
闭包 = 函数 + 它所引用的外层作用域变量。
也就是说:
被打包在了一起。
所以即使:
执行结束:
这就是闭包最核心的能力:
函数可以“记住”它创建时所在环境中的变量。
三、闭包到底“闭”住了什么?
继续看:
def outer(): name = ”小芽” def inner(): print(name) return inner
这里存在三个东西。
外层函数:
外层变量:
内部函数:
而:
使用了:
但 name 并不是 inner() 自己定义的。
这种变量在 Python 中叫:
自由变量(Free Variable)
当 Python 创建 inner 这个函数对象的时候,就会把它依赖的自由变量一起保留下来。
于是最终形成:
┌──────────────────────┐│ inner 函数 ││ ││ 函数代码 ││ + ││ name = ”小芽” │└──────────────────────┘
这就是 Closure。
四、闭包不是简单的“函数里面套函数”
这里特别容易产生一个误解:
只要函数里面定义了函数,就是闭包。
不准确。
例如:
def outer(): def inner(): print(”hello”) return inner
虽然存在嵌套函数,但 inner() 并没有引用 outer() 中的局部变量。
真正典型的闭包是:
def outer(): name = ”小芽” def inner(): print(name) return inner
因为内部函数:
引用了外层变量:
Python 因此需要为这个变量保留一个 Closure Cell。
所以更准确地说:
闭包的核心不是嵌套,而是内部函数捕获了外部作用域中的自由变量。
五、闭包最经典的用途:函数工厂
现在来看一个非常重要的概念:
Function Factory,函数工厂。
例如我们想生成不同倍数的计算函数。
可以写:
def create_multiplier(n): def multiply(x): return x * n return multiply
创建:
double = create_multiplier(2)triple = create_multiplier(3)
执行:
print(double(10))print(triple(10))
结果:
发生了什么?
第一次:
double = create_multiplier(2)
实际上创建了:
第二次:
triple = create_multiplier(3)
创建了:
所以:
double ↓记住 n = 2triple ↓记住 n = 3
同一套逻辑,通过不同外部参数,就生成了不同函数。
这正是闭包在 Agent 开发中特别有价值的原因。
六、为什么 Agent 开发特别适合使用闭包?
因为 Agent 系统中存在大量:
逻辑一样,但是运行上下文不同。
例如景区运营 Agent 里有三个景区:
都有查询订单工具:
逻辑基本一样。
区别只是:
一种非常笨的写法是:
def query_huangshan_order(order_id): ...def query_xihu_order(order_id): ...def query_jiuzhaigou_order(order_id): ...
显然不合理。
更合理的方法是:
def create_order_tool(scenic_id): def query_order(order_id): return db.query_order( scenic_id=scenic_id, order_id=order_id ) return query_order
然后:
huangshan_tool = create_order_tool(”huangshan”)xihu_tool = create_order_tool(”xihu”)jiuzhaigou_tool = create_order_tool(”jiuzhaigou”)
于是:
huangshan_tool ↓query_order + scenic_id=huangshanxihu_tool ↓query_order + scenic_id=xihujiuzhaigou_tool ↓query_order + scenic_id=jiuzhaigou
这就是:
使用闭包动态生成带上下文的 Agent Tool。
讲到这里,很多人其实会马上产生一个非常合理的疑问:
不就是参数不同吗?我直接给函数传不同参数,不也一样能查询不同的数据吗?为什么非要用闭包?
这个问题非常重要。
而且答案是:
对,如果只是从“查询结果”来看,直接传参数完全可以,闭包并没有什么神奇的能力。
例如景区运营 Agent 里有多个景区:
我们要查询不同景区的订单。
完全可以写成:
def query_order(scenic_id, order_id): return db.query_order( scenic_id=scenic_id, order_id=order_id )
调用的时候:
query_order( "huangshan", "10001")
query_order( "xihu", "10002")
当然能正常工作。
所以这里千万不要误解成:
闭包可以根据不同参数查询不同数据,而普通函数不行。
不是。
真正的区别是:
这个参数,到底由谁来传?又应该在什么时候传?
1. 普通参数:调用函数的时候再传
普通函数:
def query_order( scenic_id, order_id): ...
意味着每次调用,都需要告诉它:
例如:
query_order( scenic_id="huangshan", order_id="10001")
可以理解成:
query_order(景区, 订单号)调用的时候告诉函数:景区 = 黄山订单号 = 10001
这在普通业务开发里完全没有问题。
2. 闭包:创建函数的时候,就提前固定一部分参数
现在换成闭包:
def create_order_tool(scenic_id): def query_order(order_id): return db.query_order( scenic_id=scenic_id, order_id=order_id ) return query_order
创建:
huangshan_tool = create_order_tool( "huangshan")
这时候发生了一件事情:
已经提前绑定到了:
里面。
以后查询订单:
就不需要再传:
因为它已经记住了。
可以这样理解:
普通函数:query_order( scenic_id, order_id)两个参数都在调用时传
而闭包:
create_order_tool(”huangshan”) ↓scenic_id 提前固定 ↓huangshan_tool(order_id)
变成:
所以闭包真正改变的不是:
能不能查询不同数据。
而是:
参数什么时候绑定。
普通函数:
闭包:
3. 那这样写到底有什么实际意义?
如果只是我们自己写 Python:
query_order( "huangshan", "10001")
那么说实话:
很多情况下确实没有必要为了用闭包而用闭包。
直接传参数反而更加简单。
闭包真正开始体现价值,是进入:
Agent Tool Calling
以后。
因为 Agent 开发和普通函数调用有一个非常大的区别:
调用 Tool 的人,很可能是 LLM。
4. 关键区别来了:谁在给函数传参数?
假设我们把下面这个函数注册给 Agent:
def query_order( scenic_id: str, order_id: str): ...
那么对于 Agent 来说,这个 Tool 有两个参数:
query_order参数:scenic_idorder_id
模型调用 Tool 时,很可能生成:
{ "scenic_id": "huangshan", "order_id": "10001"}
表面看没有问题。
但是仔细想一下:
scenic_id 为什么要让 LLM 来决定?
假设当前登录系统的用户是:
系统其实早就已经知道:
等于:
那么这个参数根本不应该再让模型生成。
否则模型理论上甚至可能调用:
{ "scenic_id": "jiuzhaigou", "order_id": "10001"}
可是当前用户明明属于:
这时候问题就不仅仅是代码优不优雅了。
而可能涉及:
这时候闭包的价值就出来了。
5. 使用闭包,把 scenic_id 从 LLM 手里拿回来
系统登录以后已经知道:
scenic_id = current_user.scenic_id
于是我们创建:
def create_order_tool(scenic_id): def query_order( order_id: str ): return db.query_order( scenic_id=scenic_id, order_id=order_id ) return query_order
当前用户属于黄山:
query_order = create_order_tool( "huangshan")
现在注册给 Agent 的函数已经变成:
模型看到的参数只剩:
于是模型调用:
至于:
是谁提供的?
不是 LLM。
而是:
登录系统 ↓获取当前用户 ↓获取 scenic_id ↓创建 Tool ↓Closure 保存 scenic_id
完整流程变成:
用户登录 ↓系统知道:scenic_id = huangshan ↓create_order_tool( ”huangshan”) ↓生成:query_order(order_id) ↓注册给 Agent ↓LLM 只负责:order_id = 10001 ↓调用:query_order(”10001”) ↓Closure 自动带上:scenic_id = huangshan ↓数据库
这时候就能真正体会到闭包的价值了。
6. 普通参数和闭包的差别,不在结果,而在控制权
两种写法最终可能查询的是完全相同的数据。
普通函数:
query_order( scenic_id="huangshan", order_id="10001")
闭包:
huangshan_tool(order_id="10001")
最终进入数据库的条件可能完全一样:
WHERE scenic_id = 'huangshan'AND order_id = '10001'
所以:
闭包不是让查询功能更强。
真正的区别在于:
由谁控制。
第一种:
第二种:
系统 ↓scenic_id ↓Closure ↓Tool
因此可以总结成一句非常重要的话:
普通参数和闭包的区别,很多时候不是“能不能实现”,而是谁拥有参数的控制权。
7. 在 Agent 中,哪些参数适合让 LLM 传?
例如用户说:
那么:
确实来源于用户问题。
这种参数可以让 LLM 提取:
类似的还有:
这些属于:
业务参数。
LLM 可以决定。
8. 哪些参数不应该让 LLM 传?
例如:
tenant_iduser_idscenic_idtrace_idaccess_token数据库连接MCP Client权限信息认证信息
这些参数本质上来自:
系统运行环境。
它们并不是用户自然语言问题的一部分。
因此一般不应该出现这种 Tool:
query_order( order_id, tenant_id, user_id, trace_id)
否则 LLM 需要生成:
{ "order_id": "10001", "tenant_id": "scenic_001", "user_id": "9527", "trace_id": "abc123"}
明显不合理。
更合理的是:
而:
由程序自己维护。
闭包就是一种非常方便的实现方式。
9. 再看一个 Trace ID 的例子,就更明显了
假设每次 Agent 请求都生成一个 Trace ID:
如果 Tool 写成:
def query_order( order_id, trace_id): ...
然后注册给 Agent。
模型每次调用都要生成:
{ ”order_id”: ”10001”, ”trace_id”: ”abc123”}
这显然很奇怪。
因为:
本来就是系统自己生成的。
为什么让大模型负责填写?
所以可以:
def create_query_order_tool( trace_id): def query_order( order_id ): print( "trace:", trace_id ) return db.query_order( order_id ) return query_order
请求进来:
trace_id = create_trace_id()
然后:
query_order = ( create_query_order_tool( trace_id ))
现在:
已经自动带着:
模型只需要调用:
实际上系统内部拥有的是:
query_order+order_id = 10001+trace_id = abc123
只是:
由 LLM 提供。
而:
由系统提供。
10. 最后再回头看最开始的景区例子
最开始我们写:
def create_order_tool( scenic_id): def query_order( order_id ): return db.query_order( scenic_id=scenic_id, order_id=order_id ) return query_order
然后:
huangshan_tool = ( create_order_tool( "huangshan" ))xihu_tool = ( create_order_tool( "xihu" ))jiuzhaigou_tool = ( create_order_tool( "jiuzhaigou" ))
于是:
huangshan_tool ↓query_order+scenic_id = huangshanxihu_tool ↓query_order+scenic_id = xihujiuzhaigou_tool ↓query_order+scenic_id = jiuzhaigou
如果只是为了自己调用:
query_order( scenic_id, order_id)
当然也能完成任务。
但进入 Agent 系统以后,我们往往希望做到:
LLM 只控制:order_iddateshop_namekeyword查询条件系统自己控制:tenant_idscenic_iduser_idtrace_id权限认证信息数据库连接MCP Client
于是闭包就非常适合承担:
把系统上下文提前绑定到 Tool 中。
最终可以把这一节浓缩成一句话:
闭包并不是为了替代函数传参,而是为了把“不应该由调用者每次提供的参数”提前绑定到函数中。
尤其是在 Agent 开发里:
模型负责业务参数,系统负责安全和运行上下文。
这才是闭包在 Agent Tool 设计里真正重要的价值。
七、Agent 场景一:给 Tool 注入依赖
这是闭包在 Agent 开发中非常典型的用途。
假设我们有:
class OrderService: def get_order(self, order_id): ...
Agent Tool 需要调用它。
可以写:
def create_order_tool(order_service): def query_order(order_id: str): return order_service.get_order(order_id) return query_order
初始化:
order_service = OrderService()query_order = create_order_tool(order_service)
以后 Agent 只需要调用:
query_order(”202608190001”)
而不需要每次传:
因为:
query_order已经通过闭包记住了order_service
从 Agent 的角度:
Agent ↓query_order(order_id)
从系统内部看:
query_order ↓Closure ↓OrderService ↓数据库 / RPC / MCP
这里其实实现了一种非常轻量的:
依赖注入。
八、Agent 场景二:创建带配置的 Tool
假设你的 Agent 同时连接:
每个 MCP Server:
我们可以写:
def create_mcp_tool(base_url, token, timeout=10): async def call_tool(tool_name, arguments): return await request( url=f"{base_url}/tools/{tool_name}", token=token, timeout=timeout, json=arguments ) return call_tool
初始化:
cashier_mcp = create_mcp_tool( "https://cashier.example.com", "cashier-token")food_mcp = create_mcp_tool( "https://food.example.com", "food-token")
于是:
cashier_mcp ↓记住 cashier URL记住 cashier Token记住 timeoutfood_mcp ↓记住 food URL记住 food Token记住 timeout
真正调用的时候只关心:
await cashier_mcp( ”query_order”, {”order_id”: ”10001”})
而不是:
await call_tool( base_url=..., token=..., timeout=..., tool_name=..., arguments=...)
这就是闭包非常重要的一个价值:
把初始化阶段的配置和真正业务调用阶段的参数分离。
九、Agent 场景三:动态生成 LangGraph Node
这个场景特别常见。
假设有三个模型:
每个 LangGraph Node 都是类似逻辑:
async def node(state): return await model.ainvoke(state[”messages”])
如果分别写:
async def analysis_node(state): return await analysis_model.ainvoke(state[”messages”])async def summary_node(state): return await summary_model.ainvoke(state[”messages”])async def plan_node(state): return await plan_model.ainvoke(state[”messages”])
会有很多重复代码。
可以改成:
def create_llm_node(model): async def node(state): response = await model.ainvoke( state[”messages”] ) return { ”messages”: [response] } return node
然后:
analysis_node = create_llm_node(analysis_model)summary_node = create_llm_node(summary_model)plan_node = create_llm_node(plan_model)
这里每一个 Node:
组成一个闭包。
然后放进 LangGraph:
graph.add_node( "analysis", analysis_node)graph.add_node( "summary", summary_node)graph.add_node( "plan", plan_node)
这样节点本身不需要:
LangGraph 仍然只需要按照它要求的函数签名调用:
但是内部实际上已经绑定好了模型。
这一点非常重要。
十、闭包为什么特别适合框架回调?
很多框架会规定:
你的函数只能接收固定参数。
例如框架要求:
但你的业务实际上还需要:
modeldatabaseredistenant_idtrace_servicepermission_service
你又不能把函数写成:
node( state, model, database, redis, tenant_id, trace_service)
因为框架根本不会这样调用你。
这时候闭包就非常有用了。
你可以:
def create_node( model, database, redis, tenant_id): async def node(state): # 可以使用: # model # database # redis # tenant_id ... return node
最终框架看到的仍然只是:
但实际上它背后已经绑定:
modeldatabaseredistenant_id
所以可以把闭包理解成:
给固定函数签名偷偷挂上一份运行上下文。
这里的“偷偷”只是帮助理解,工程上更准确的说法是:
通过词法作用域捕获依赖。
十一、Agent 场景四:绑定租户和用户上下文
企业 Agent 很容易遇到多租户问题。
例如:
Agent 查询订单时绝对不能发生:
一种方式就是给 Tool 绑定 Tenant Context。
例如:
def create_order_tool( order_service, tenant_id): async def query_order(order_id: str): return await order_service.query( tenant_id=tenant_id, order_id=order_id ) return query_order
创建:
tool = create_order_tool( order_service, tenant_id=”scenic_001”)
Agent 后续只能:
而不需要让模型决定:
这是一个非常值得注意的设计。
不要让 LLM 调用:
query_order( tenant_id="???", order_id="...")
因为:
属于系统安全上下文,不应该让模型随便填写。
更好的设计是:
HTTP 登录用户 ↓获得 tenant_id ↓创建当前请求 Tool ↓Closure 捕获 tenant_id ↓LLM 只能提供 order_id
即:
而不是:
query_order(tenant_id, order_id)
这就是:
利用闭包缩小 Tool 暴露给模型的参数面。
这在 Agent Tool 设计里非常实用。
十二、Agent 场景五:权限控制
假设 Agent 有退款工具:
但是:
普通员工:不能退款主管:退款 ≤ 500 元经理:退款 ≤ 5000 元
可以创建一个权限包装器:
def with_permission( tool, permission_service, user_id): async def wrapped(*args, **kwargs): allowed = await permission_service.check( user_id=user_id, tool_name=tool.__name__ ) if not allowed: raise PermissionError( "当前用户无权限执行该操作" ) return await tool( *args, **kwargs ) return wrapped
然后:
safe_refund_tool = with_permission( refund_order, permission_service, user_id)
Agent 调用:
await safe_refund_tool( order_id="10001")
实际上执行流程:
Agent ↓safe_refund_tool ↓检查 user_id 权限 ↓允许? ├── 否 → 拒绝 └── 是 ↓ refund_order
这里:
toolpermission_serviceuser_id
全部被闭包保存。
十三、你会发现:这已经很像装饰器了
没错。
Python 的装饰器,本质上和闭包有非常密切的关系。
例如:
def log_tool(func): async def wrapper(*args, **kwargs): print( f"开始调用工具: {func.__name__}" ) result = await func( *args, **kwargs ) print( f"工具调用完成: {func.__name__}" ) return result return wrapper
使用:
@log_toolasync def query_order(order_id): ...
实际上:
query_order = log_tool(query_order)
新的:
内部保存了原来的:
所以:
装饰器经常就是建立在闭包之上的。
闭包解决:
装饰器解决:
十四、Agent 场景六:Tool 调用日志和 Trace
企业 Agent 一定需要可观测性。
例如:
用户问题 ↓Agent ↓Tool Call ↓MCP ↓数据库
你可能希望记录:
trace_iduser_idtool_namedurationsuccesserror
可以写:
import timedef create_traced_tool( tool, trace_id, trace_service): async def wrapped( *args, **kwargs ): start = time.time() try: result = await tool( *args, **kwargs ) await trace_service.record( trace_id=trace_id, tool=tool.__name__, duration=time.time() - start, success=True ) return result except Exception as e: await trace_service.record( trace_id=trace_id, tool=tool.__name__, duration=time.time() - start, success=False, error=str(e) ) raise return wrapped
这里:
tooltrace_idtrace_service
都通过闭包绑定。
最终 Agent 完全不用关心 Trace:
但是实际上:
query_order() ↓Trace Wrapper ↓真实 Tool
这就是一个很典型的:
横切能力。
类似能力还包括:
这些都非常适合:
实现。
十五、Agent 场景七:重试策略
例如调用 MCP Server:
可能偶尔出现:
Timeout503ConnectionError
我们希望给不同 Tool 配不同的 Retry。
可以写:
import asynciodef with_retry( tool, max_retries=3, delay=1): async def wrapped( *args, **kwargs ): for attempt in range(max_retries): try: return await tool( *args, **kwargs ) except Exception: if attempt == max_retries - 1: raise await asyncio.sleep(delay) return wrapped
然后:
query_order = with_retry( query_order, max_retries=3)
库存接口:
query_inventory = with_retry( query_inventory, max_retries=5)
最终:
同样的 retry 逻辑 ↓通过不同闭包保存不同配置query_ordermax_retries = 3query_inventorymax_retries = 5
这再次体现了:
闭包非常适合“逻辑相同、配置不同”的场景。
十六、Agent 场景八:创建 Callback
Agent 系统里还有大量回调:
on_agent_starton_agent_endon_tool_starton_tool_endon_llm_starton_llm_end
假设我们想创建一个 Tool 回调:
def create_tool_callback(trace_id): async def on_tool_end( tool_name, result ): print( trace_id, tool_name, result ) return on_tool_end
创建:
callback = create_tool_callback( ”trace_10001”)
以后框架只负责调用:
await callback( ”query_order”, result)
但是 callback 自己知道:
这就是典型的:
事件处理函数 + 上下文
而这正是闭包特别适合解决的问题。
十七、闭包还可以保存状态
到目前为止,我们保存的是:
闭包实际上还可以保存:
状态。
例如统计 Tool 调用了多少次。
def create_counter(): count = 0 def increment(): nonlocal count count += 1 return count return increment
使用:
counter = create_counter()print(counter())print(counter())print(counter())
输出:
为什么?
因为:
被 Closure 保存了下来。
十八、这里为什么需要 nonlocal?
看:
def create_counter(): count = 0 def increment(): count += 1 return count return increment
这样会报错。
因为:
相当于:
Python 看到:
会认为:
count 是 increment() 的局部变量
但是右边又要先读取:
于是就会报错。
需要告诉 Python:
意思是:
这个 count 不是当前函数的局部变量,而是使用外层函数中的 count。
所以:
def create_counter(): count = 0 def increment(): nonlocal count count += 1 return count return increment
十九、Agent 中能不能用闭包保存状态?
可以。
例如:
def create_tool_monitor(): success_count = 0 error_count = 0 async def execute(tool, *args, **kwargs): nonlocal success_count nonlocal error_count try: result = await tool( *args, **kwargs ) success_count += 1 return result except Exception: error_count += 1 raise return execute
理论上可以维护:
但是这里需要特别提醒:
企业级 Agent 开发中,不要过度使用闭包保存重要业务状态。
例如这些东西:
Agent SessionConversation State订单状态任务执行状态审批状态长期 MemoryWorkflow State
不要因为闭包可以保存状态,就全部塞进 Closure。
这些状态更适合:
LangGraph StateRedisPostgreSQL数据库CheckpointSession StoreMemory Store
原因很简单。
闭包状态通常是:
如果:
状态就可能丢失或者不一致。
所以企业开发里更推荐:
闭包主要保存轻量级上下文和依赖Redis / DB / State保存真正业务状态
二十、一个非常重要的问题:并发安全
Agent 后端通常是:
这时候不要随便创建全局闭包状态。
例如:
然后整个 FastAPI 应用所有用户共享:
User AUser BUser C ↓同一个 counter
就可能发生状态串扰。
尤其下面这些内容:
user_idtenant_idtrace_idconversation_idrequest_id
如果错误地放进全局共享 Closure,就可能导致严重问题。
因此需要区分:
例如这种:
query_tool = create_tool( db_client)
db_client 本身支持并发,而且所有请求共享,这是合理的。
但是:
query_tool = create_tool( tenant_id=”scenic_001”)
如果整个应用都使用这个工具,就不一定合理。
更常见的是:
请求进入 ↓身份认证 ↓获取 user_idtenant_idtrace_id ↓创建当前请求上下文 ↓生成 Request Scoped Tool ↓交给 Agent
这样每个请求拥有自己的:
不会相互污染。
二十一、Agent 场景九:动态生成一组工具
例如景区 Agent 根据用户身份允许访问不同系统:
运营人员├── 查询订单├── 查询营收└── 查询库存财务人员├── 查询订单├── 查询支付├── 查询退款└── 对账管理员├── 查询├── 修改├── 退款└── 配置
可以在请求开始时:
def build_tools( user, services): tools = [] if user.can_query_order: async def query_order(order_id): return await services.order.query( tenant_id=user.tenant_id, order_id=order_id ) tools.append(query_order) if user.can_refund: async def refund_order(order_id): return await services.order.refund( tenant_id=user.tenant_id, order_id=order_id ) tools.append(refund_order) return tools
最终:
当前用户 ↓权限系统 ↓动态创建 Tool Set ↓交给 Agent
因此不同用户看到的 Tool 根本不一样。
这是比单纯告诉 LLM:
安全得多的方案。
因为:
没有权限的 Tool,压根就不要注册给模型。
二十二、闭包和 Agent Tool Factory
现在可以正式引出一个非常实用的设计模式:
Tool Factory。
例如:
def create_order_tool( order_service, tenant_id, user_id, trace_id): async def query_order( order_id: str ): return await order_service.query( tenant_id=tenant_id, user_id=user_id, trace_id=trace_id, order_id=order_id ) return query_order
这时候:
系统内部上下文:order_servicetenant_iduser_idtrace_id ↓Tool Factory ↓LLM Tool:query_order(order_id)
这实际上完成了一件非常漂亮的事情:
把“系统参数”和“模型参数”分开。
系统参数:
tenant_iduser_idtrace_idservice权限认证信息
由程序控制。
模型参数:
order_iddateshop_namequery_condition
由 LLM 根据用户问题提取。
这是企业 Agent Tool 设计中非常重要的边界。
二十三、来看一个完整的景区 Agent 示例
假设用户问:
HTTP 请求进入:
async def chat(request): user = request.user trace_id = create_trace_id() tools = build_agent_tools( user=user, trace_id=trace_id ) agent = create_agent( model=model, tools=tools ) return await agent.ainvoke( request.message )
Tool Builder:
def build_agent_tools( user, trace_id): def create_query_order_tool(): tenant_id = user.tenant_id user_id = user.id async def query_order( order_id: str ): return await order_service.query( tenant_id=tenant_id, user_id=user_id, trace_id=trace_id, order_id=order_id ) return query_order return [ create_query_order_tool() ]
最终模型实际看到的 Tool Schema 可能只有:
模型根本不知道:
tenant_iduser_idtrace_id数据库连接认证 Token
这些东西都由应用层管理。
完整结构就是:
用户请求 ↓FastAPI ↓Authentication ↓user_idtenant_idtrace_id ↓Tool Factory ↓Closure ↓query_order(order_id) ↓注册给 Agent ↓LLM ↓Tool Calling ↓query_order(”202608190001”) ↓Closure 自动补充tenant_iduser_idtrace_id ↓Order Service ↓数据库 / MCP
理解到这里,你就已经不是在学习一个单纯的 Python 语法点了。
你是在理解:
Agent Runtime 中上下文是怎么注入执行函数的。
二十四、闭包和普通参数传递有什么区别?
可能有人会问:
为什么不直接:
query_order( order_id, tenant_id, user_id, trace_id)
当然可以。
但问题在于:
Agent Tool Schema 会变成:
{ ”order_id”: ”...”, ”tenant_id”: ”...”, ”user_id”: ”...”, ”trace_id”: ”...”}
这意味着 LLM 需要生成:
这几个参数根本不应该由 LLM 决定。
所以:
明显更合理。
外部:
内部:
二十五、闭包和类有什么区别?
闭包能保存状态,类也能。
例如闭包:
def create_tool(client): async def tool(order_id): return await client.query(order_id) return tool
也可以写成:
class OrderTool: def __init__(self, client): self.client = client async def __call__(self, order_id): return await self.client.query(order_id)
使用:
tool = OrderTool(client)await tool(”10001”)
两者本质上都可以实现:
怎么选?
如果只是:
绑定几个参数轻量级包装简单回调简单 Factory
闭包通常非常舒服。
例如:
modeltenant_idtrace_idtimeoutretry_count
但是如果对象开始拥有:
很多状态很多方法复杂生命周期复杂继承关系需要明确接口
就更适合 Class。
可以简单记:
简单行为 + 少量上下文 ↓Closure复杂对象 + 多种行为 + 生命周期 ↓Class
二十六、闭包和 lambda 是什么关系?
闭包和 lambda 不是一个概念。
lambda 只是:
匿名函数语法。
例如:
也可以形成闭包:
def create_multiplier(n): return lambda x: x * n
这里:
引用了:
所以它也是闭包。
因此:
lambda解决的是匿名函数Closure解决的是函数捕获外部变量
两者可以组合,但不是同一个东西。
二十七、Agent 开发中一个非常经典的闭包坑:Late Binding
这个坑一定要知道。
假设我们动态创建 Tool:
tools = []for scenic_id in [ "scenic_001", "scenic_002", "scenic_003"]: def query(): return scenic_id tools.append(query)
然后:
print(tools[0]())print(tools[1]())print(tools[2]())
你可能以为是:
scenic_001scenic_002scenic_003
实际上很可能得到:
scenic_003scenic_003scenic_003
为什么?
因为 Python 闭包捕获的不是:
而是:
等循环结束以后:
scenic_id == ”scenic_003”
所以所有函数读取的都是:
二十八、怎么解决 Late Binding?
最容易理解的方法就是:
再加一层函数工厂。
例如:
def create_query_tool(scenic_id): def query(): return scenic_id return query
然后:
tools = []for scenic_id in scenic_ids: tools.append( create_query_tool(scenic_id) )
这样每一次:
都会创建一个新的作用域。
得到:
Tool 1 ↓scenic_id = 001Tool 2 ↓scenic_id = 002Tool 3 ↓scenic_id = 003
这就是为什么在 Agent 代码里,你经常会看到:
这样的 Factory 函数。
不仅仅是为了“代码优雅”。
它还天然创建了:
独立的 Closure Scope。
二十九、为什么 Agent 框架源码里到处都是闭包?
现在再去看:
LangChainLangGraphFastAPI各种 Agent Framework
你经常会发现类似代码:
def create_handler(config): async def handler(state): ... return handler
或者:
def middleware(config): def wrapper(func): async def wrapped(*args, **kwargs): ... return wrapped return wrapper
以前看起来可能会觉得:
怎么套这么多层函数?
实际上每一层都可能是在做:
绑定 Config绑定 Runtime绑定 Callback绑定 Client绑定 Retry绑定 Context绑定 Permission绑定 Trace
最终外部框架只需要看到一个很干净的:
内部却已经携带了一整套执行上下文。
三十、从 Agent 架构角度重新理解闭包
如果只从 Python 语法角度学习,闭包可能是:
这个定义没错。
但是站在 Agent 工程角度看,可以把它理解得更深一点:
Closure 是一种轻量级的上下文绑定机制。
例如:
Agent │ ↓ query_order() │ ↓ ┌───────────┐ │ Closure │ ├───────────┤ │ tenant_id │ │ user_id │ │ trace_id │ │ service │ │ config │ └───────────┘ │ ↓ OrderService │ ↓ MCP / DB
模型看到的是:
系统内部拥有的是:
query_order+用户上下文+租户上下文+服务依赖+Trace 信息+权限信息
所以 Closure 很适合解决:
执行逻辑和执行上下文如何绑定。
三十一、Agent 开发里,闭包最常见的 8 个用途
现在可以总结一下。
1. Tool Factory
动态生成工具:
2. Node Factory
动态生成 LangGraph Node:
3. Dependency Injection
绑定:
DatabaseRedisMCP ClientServiceLLMRetriever
4. Request Context
绑定:
user_idtenant_idtrace_idconversation_id
5. Callback
创建:
on_tool_starton_tool_endon_llm_end
等回调。
6. Middleware / Decorator
实现:
7. Config Binding
绑定:
timeoutretrymodeltemperaturetop_kserver_url
8. Lightweight State
维护少量:
但不建议拿它保存关键业务状态。
三十二、最后把闭包、Tool、Agent 串起来
假设你的景区智能运营平台用户问:
系统获取:
tenant_id = scenic_001user_id = 9527trace_id = abc123
然后创建工具:
def create_order_tool( service, tenant_id, user_id, trace_id): async def query_order( order_id: str ): return await service.query( order_id=order_id, tenant_id=tenant_id, operator=user_id, trace_id=trace_id ) return query_order
生成:
query_order = create_order_tool( order_service, "scenic_001", "9527", "abc123")
此时可以把:
理解成:
query_order│├── 函数代码│├── service│├── tenant_id = scenic_001│├── user_id = 9527│└── trace_id = abc123
然后注册给 Agent:
Agent ↓query_order(order_id)
LLM 只需要生成:
剩下:
哪个景区?哪个用户?哪个 Trace?调用哪个 Service?
完全不用交给 LLM。
这就是闭包真正有价值的地方。
三十三、一句话理解闭包
如果只为了 Python 面试,可以记:
闭包是一个函数以及它所捕获的外层作用域自由变量组成的结构,使外层函数执行结束后,内部函数仍然能够访问这些变量。
但如果是做 Agent 开发,我更建议记住下面这句话:
闭包就是给一个函数绑定一份上下文,让函数以后无论在哪里被调用,都还能使用创建它时绑定的配置、依赖和环境。
而 Agent 开发恰好到处都是:
例如:
Tool+Tenant ContextNode+LLMCallback+Trace ContextHandler+ConfigTool+ServiceWrapper+Permission
所以闭包并不是 Agent 独有的技术。
但是:
Agent 的 Tool、Node、Callback、Middleware、Factory、Runtime Context 这些设计,让闭包变得特别常用。
以后再看到这种代码:
def create_xxx(config): async def xxx(state): ... return xxx
不要只看到:
函数里面套函数。
你应该马上意识到:
create_xxx() ↓正在创建一个执行函数 +给它绑定运行上下文 ↓形成 Closure ↓交给 Agent / LangGraph / Callback 系统
理解到这一层,你才算真正知道:
为什么 Python Agent 项目里经常出现闭包。