第 4 章 函数一等公民与原生元编程
4.1 lambda 与函数是一等公民
square=lambdax:x*x# 只能单表达式，能写 def 就别写 lambda# 真正的用途：当参数传users.sort(key=lambdau:u["age"])# ↔ Comparator.comparingmax(items,key=lambdax:x["priority"])functools.reduce(lambdaa,b:a+b,nums,0)与 Java 的差异：所有函数都是一等公民，不需要Function<T,R>/Supplier<T>这些函数式接口的载体——函数直接传、直接赋值、直接存进 dict：
# 用字典代替 switch/if-else 链 —— handler 注册表，agent 工具路由的惯用形态handlers:dict[str,Callable[[dict],str]]={"query":handle_query,"book":handle_booking,"cancel":handle_cancel,}result=handlers.get(intent,handle_unknown)(payload)# 查表调用类型标注用Callable：
fromtypingimportCallable ToolFunc=Callable[[dict],str]# 参数 dict、返回 str 的可调用对象Callback=Callable[...,None]# 参数不定4.2 装饰器 ↔ 注解 + AOP
Java 注解（@Override、@Transactional）本身只是元数据，需要框架反射处理。Python 装饰器则是直接执行的函数包装：@deco就是f = deco(f)的语法糖，无框架、无反射，运行时立即生效。
importfunctools,timedeftimed(func):"""记录执行耗时——等价于手写 AOP 环绕通知。"""@functools.wraps(func)# 保留原函数名/文档（惯例必写）defwrapper(*args,**kwargs):start=time.perf_counter()result=func(*args,**kwargs)print(f"{func.__name__}:{time.perf_counter()-start:.2f}s")returnresultreturnwrapper@timed# ↔ timed(call_llm)defcall_llm(prompt:str)->str:...类内置装饰器三件套 + 标准库常用装饰器：
classRegistry:instance:"Registry | None"=None@staticmethoddefutils()->str:...# ↔ 静态方法（无 self）@classmethoddefget_instance(cls)->"Registry":# ↔ 工厂方法，cls 即类本身...@propertydefsize(self)->int:...# 见 3.3@functools.cache# 标准库缓存 ↔ Guava Cache（无淘汰）defexpensive(self,key:str)->str:...带参数的装饰器（三层嵌套，实际多用functools.wraps模板）：
defretry(max_attempts:int=3,delay:float=1.0):"""agent 工具调用失败重试——生产代码真实写法。"""defdecorator(func):@functools.wraps(func)defwrapper(*args,**kwargs):forattemptinrange(1,max_attempts+1):try:returnfunc(*args,**kwargs)exceptException:ifattempt==max_attempts:raisetime.sleep(delay*attempt)returnwrapperreturndecorator@retry(max_attempts=3,delay=0.5)defcall_search_api(query:str)->str:...Agent 框架中常见的高频装饰器：
frompydantic_aiimportAgent# FastAPI 路由、LangGraph 节点、MCP server 的工具注册，全是装饰器：@app.get("/chat")# FastAPI：注册路由@mcp.tool()# MCP：把函数注册为工具@agent.tool# PydanticAI：注册 agent 工具defget_weather(city:str)->str:...闭包作用域链 LEGB 与 nonlocal 关键字避坑
在 Java 中，Lambda 或匿名内部类捕获外层局部变量时，编译器强制要求变量必须是effectively final（只读，不可修改）。
Python 允许闭包捕获外层变量，其变量查找遵循LEGB 规则：Local（局部） ➔ Enclosing（闭包外层函数） ➔ Global（模块全局） ➔ Builtin（内置名称）。
闭包内直接赋值引发UnboundLocalError：
defcreate_counter():count=0definc():# count += 1 等价于 count = count + 1# Python 解释器在编译期扫描到赋值符号 '='，将 count 判定为 inc 函数内的 Local 变量！# 结果在执行 count + 1 时，读取了尚未初始化的 Local 变量：count+=1# 运行时抛出 UnboundLocalError: local variable 'count' referenced before assignmentreturncountreturninc标准解法：使用nonlocal显式声明：
defcreate_counter():count=0definc():nonlocalcount# 显式声明：这是 Enclosing 作用域的变量，不是新创建的局部变量count+=1returncountreturninc# 在手写有状态装饰器（如统计调用次数）时的标准应用：defcall_tracker(func):calls=0@functools.wraps(func)defwrapper(*args,**kwargs):nonlocalcalls calls+=1print(f"[{func.__name__}] 已被调用{calls}次")returnfunc(*args,**kwargs)returnwrapper若需要修改模块顶层的全局变量，则在函数内使用global var_name。
4.3 上下文管理器 with ↔ try-with-resources
// Javatry(varconn=pool.getConnection()){conn.query(sql);}// 自动 close()withopen("config.json",encoding="utf-8")asf:# ↔ try-with-resourcesconfig=json.load(f)# 退出 with 块时自动调用 f.close()，即使抛异常# 一次管理多个资源withpool.connection()asconn,open("log.txt","a")asf:...任何对象实现__enter__/__exit__即可支持 with。agent 开发常用场景：数据库会话、trace span、临时目录、文件句柄。自己写上下文管理器用contextlib更快：
fromcontextlibimportcontextmanager@contextmanagerdeftimer(label:str):# 生成器函数版上下文管理器start=time.perf_counter()try:yield# yield 前=__enter__，后=__exit__finally:print(f"{label}:{time.perf_counter()-start:.3f}s")withtimer("llm call"):llm.invoke(prompt)# 常用内置/库提供的：open、sqlalchemy session、unittest.mock、# torch.no_grad()（如有 ML 需求）4.4 生成器与迭代器：yield（重点）
yield让函数变成生成器：调用时不执行，每次取值时执行到下一个 yield。这是 Python 版的"惰性 Stream"，也是LLM 流式输出的实现基础——所有 SDK 的流式返回都是生成器。
defread_lines(path:str):withopen(path,encoding="utf-8")asf:forlineinf:line=line.strip()ifline:yieldline# 每次产出一条，读完即停，不占内存forlineinread_lines("big.log"):# 逐条消费process(line)// Java 最近的等价物Stream<String>lines=Files.lines(path);// 惰性，但要手动 close对比 Java Stream：生成器是** pull 模型**（消费者驱动），且可以yield任意次、有状态。理解 agent 流式的关键：
# 典型：LLM 流式输出（openai SDK 真实返回形态）defstream_chat(messages:list[dict])->Iterator[str]:stream=client.chat.completions.create(model="gpt-4o",messages=messages,stream=True,)forchunkinstream:# 网络上每到达一段就产出一段ifchunk.choices[0].delta.content:yieldchunk.choices[0].delta.content# 上层直接透传给 SSEdefsse_endpoint():yieldfromstream_chat(history)# yield from ↔ 委托给另一个生成器类型标注：Iterator[str]/Generator[str, None, None]。
三个必须掌握的衍生点：
# 1. 生成器表达式 —— 推导式的惰性版，适合大集合total=sum(len(m)forminmessages)# 惰性，边遍历边算，无中间 listfirst=next((mforminmessagesifm["role"]=="tool"),None)# ↔ findFirst# 2. 一次性！生成器只能消费一遍，消费完为空 —— Stream 同样如此g=(xforxin[1,2,3])list(g)# [1, 2, 3]list(g)# []# 3. 无限生成器 + islice 截断fromitertoolsimportislicedefretry_forever(fn):# agent 重试循环常用形态whileTrue:try:yieldfn()exceptExceptionase:print("retry:",e)forresultinislice(retry_forever(call_api),5):# 最多 5 次...4.5 异常处理
Python 只有 unchecked exception 一种，语法更简单：
try:result=call_tool(name,args)exceptToolNotFoundErrorase:# ↔ catch (ToolNotFoundException e)logger.warning(f"tool missing:{e}")returnfallback_responseexcept(TimeoutError,ConnectionError)ase:# 多类型：元组写法returnf"transient error:{e}"exceptExceptionase:# 兜底 ↔ catch (Exception e)raise# ↔ 直接重新抛出（或裸 raise 保留原栈）else:# try 无异常时执行（Java 没有）log_success(result)finally:# 与 Java 相同cleanup()要点与差异：
- 没有 checked exception。无需声明
throws，编译器也不会强制要求捕获。调用可能失败的函数不会收到编译期提示，防御责任由编译器转移到了调用方逻辑、单元测试与类型系统。 raise关键字对应throw；裸raise（不带对象）在 except 块内重新抛出当前异常并保留原始 traceback，等价于 Java 的throw e但栈信息更完整。- 自定义异常就是继承
Exception：
classAgentError(Exception):"""agent 基础异常。"""classToolExecutionError(AgentError):def__init__(self,tool_name:str,cause:str):super().__init__(f"tool '{tool_name}' failed:{cause}")self.tool_name=tool_name# 使用try:...exceptAgentErrorase:...# 多态捕获，与 Java 一致- 异常信息直接用 f-string 组装；
e.args、str(e)取消息。 - 不存在 Java 的 “swallow or declare” 纠结，社区规范是：能处理就处理，不能处理就让它抛（fail fast），在最外层（如 FastAPI 全局 handler）统一兜底。
raise ... from e↔ 异常链，保留"根本原因"（Java 默认行为，Python 要显式写）：
try:json.loads(raw)exceptjson.JSONDecodeErrorase:raiseAgentError(f"invalid LLM output:{raw[:100]}")frome异常层次与任务取消保护（BaseException vs Exception）
在 Java 中，所有异常派生自Throwable，业务代码通常捕获Exception。Python 的异常继承树设计有着极为特殊的工程边界：
BaseException ├── SystemExit (解释器退出信号，如 sys.exit()) ├── KeyboardInterrupt (Ctrl+C 中断信号) ├── GeneratorExit (生成器关闭信号) └── Exception (所有常规业务异常的基类) ├── StandardError / ValueError / KeyError ... └── asyncio.CancelledError (Python 3.8+ 调整为继承自 BaseException)异步任务取消信号被吞的问题：
在 Agent 异步后端中，如果客户端中断连接或请求超时，事件循环会通过向协程注入asyncio.CancelledError来取消任务。
# 错误示范：随手写了 except BaseException 或裸 except:try:awaitcall_agent_step()exceptBaseExceptionase:# 吞掉了 CancelledError！logger.error("发生错误，继续兜底")# 导致任务无法被正常取消，后台协程形成僵尸死循环，持续浪费 GPU/网络资源！核心防范准则：
- 业务逻辑兜底捕获必须严格限定在
except Exception as e:，绝不捕获BaseException； - 如果确需在底层捕获所有异常进行日志记录，必须将
asyncio.CancelledError重新向上抛出：try:awaitbackground_job()exceptasyncio.CancelledError:logger.info("任务正常被取消，执行资源清理...")raise# 必须重新抛出，让事件循环感知取消exceptExceptionase:logger.exception(f"业务执行失败:{e}")
