Java 转 Python 后端 03: 函数一等公民、三层装饰器与流式生成器

第 4 章 函数一等公民与原生元编程4.1 lambda 与函数是一等公民squarelambdax:x*x# 只能单表达式能写 def 就别写 lambda# 真正的用途当参数传users.sort(keylambdau:u[age])# ↔ Comparator.comparingmax(items,keylambdax:x[priority])functools.reduce(lambdaa,b:ab,nums,0)与 Java 的差异所有函数都是一等公民不需要FunctionT,R/SupplierT这些函数式接口的载体——函数直接传、直接赋值、直接存进 dict# 用字典代替 switch/if-else 链 —— handler 注册表agent 工具路由的惯用形态handlers:dict[str,Callable[[dict],str]]{query:handle_query,book:handle_booking,cancel:handle_cancel,}resulthandlers.get(intent,handle_unknown)(payload)# 查表调用类型标注用CallablefromtypingimportCallable ToolFuncCallable[[dict],str]# 参数 dict、返回 str 的可调用对象CallbackCallable[...,None]# 参数不定4.2 装饰器 ↔ 注解 AOPJava 注解Override、Transactional本身只是元数据需要框架反射处理。Python 装饰器则是直接执行的函数包装deco就是f deco(f)的语法糖无框架、无反射运行时立即生效。importfunctools,timedeftimed(func):记录执行耗时——等价于手写 AOP 环绕通知。functools.wraps(func)# 保留原函数名/文档惯例必写defwrapper(*args,**kwargs):starttime.perf_counter()resultfunc(*args,**kwargs)print(f{func.__name__}:{time.perf_counter()-start:.2f}s)returnresultreturnwrappertimed# ↔ timed(call_llm)defcall_llm(prompt:str)-str:...类内置装饰器三件套 标准库常用装饰器classRegistry:instance:Registry | NoneNonestaticmethoddefutils()-str:...# ↔ 静态方法无 selfclassmethoddefget_instance(cls)-Registry:# ↔ 工厂方法cls 即类本身...propertydefsize(self)-int:...# 见 3.3functools.cache# 标准库缓存 ↔ Guava Cache无淘汰defexpensive(self,key:str)-str:...带参数的装饰器三层嵌套实际多用functools.wraps模板defretry(max_attempts:int3,delay:float1.0):agent 工具调用失败重试——生产代码真实写法。defdecorator(func):functools.wraps(func)defwrapper(*args,**kwargs):forattemptinrange(1,max_attempts1):try:returnfunc(*args,**kwargs)exceptException:ifattemptmax_attempts:raisetime.sleep(delay*attempt)returnwrapperreturndecoratorretry(max_attempts3,delay0.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内置名称。闭包内直接赋值引发UnboundLocalErrordefcreate_counter():count0definc():# count 1 等价于 count count 1# Python 解释器在编译期扫描到赋值符号 将 count 判定为 inc 函数内的 Local 变量# 结果在执行 count 1 时读取了尚未初始化的 Local 变量count1# 运行时抛出 UnboundLocalError: local variable count referenced before assignmentreturncountreturninc标准解法使用nonlocal显式声明defcreate_counter():count0definc():nonlocalcount# 显式声明这是 Enclosing 作用域的变量不是新创建的局部变量count1returncountreturninc# 在手写有状态装饰器如统计调用次数时的标准应用defcall_tracker(func):calls0functools.wraps(func)defwrapper(*args,**kwargs):nonlocalcalls calls1print(f[{func.__name__}] 已被调用{calls}次)returnfunc(*args,**kwargs)returnwrapper若需要修改模块顶层的全局变量则在函数内使用global var_name。4.3 上下文管理器 with ↔ try-with-resources// Javatry(varconnpool.getConnection()){conn.query(sql);}// 自动 close()withopen(config.json,encodingutf-8)asf:# ↔ try-with-resourcesconfigjson.load(f)# 退出 with 块时自动调用 f.close()即使抛异常# 一次管理多个资源withpool.connection()asconn,open(log.txt,a)asf:...任何对象实现__enter__/__exit__即可支持 with。agent 开发常用场景数据库会话、trace span、临时目录、文件句柄。自己写上下文管理器用contextlib更快fromcontextlibimportcontextmanagercontextmanagerdeftimer(label:str):# 生成器函数版上下文管理器starttime.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,encodingutf-8)asf:forlineinf:lineline.strip()ifline:yieldline# 每次产出一条读完即停不占内存forlineinread_lines(big.log):# 逐条消费process(line)// Java 最近的等价物StreamStringlinesFiles.lines(path);// 惰性但要手动 close对比 Java Stream生成器是** pull 模型**消费者驱动且可以yield任意次、有状态。理解 agent 流式的关键# 典型LLM 流式输出openai SDK 真实返回形态defstream_chat(messages:list[dict])-Iterator[str]:streamclient.chat.completions.create(modelgpt-4o,messagesmessages,streamTrue,)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. 生成器表达式 —— 推导式的惰性版适合大集合totalsum(len(m)forminmessages)# 惰性边遍历边算无中间 listfirstnext((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:resultcall_tool(name,args)exceptToolNotFoundErrorase:# ↔ catch (ToolNotFoundException e)logger.warning(ftool missing:{e})returnfallback_responseexcept(TimeoutError,ConnectionError)ase:# 多类型元组写法returnftransient 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但栈信息更完整。自定义异常就是继承ExceptionclassAgentError(Exception):agent 基础异常。classToolExecutionError(AgentError):def__init__(self,tool_name:str,cause:str):super().__init__(ftool {tool_name} failed:{cause})self.tool_nametool_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(finvalid LLM output:{raw[:100]})frome异常层次与任务取消保护BaseException vs Exception在 Java 中所有异常派生自Throwable业务代码通常捕获Exception。Python 的异常继承树设计有着极为特殊的工程边界BaseException ├── SystemExit (解释器退出信号如 sys.exit()) ├── KeyboardInterrupt (CtrlC 中断信号) ├── GeneratorExit (生成器关闭信号) └── Exception (所有常规业务异常的基类) ├── StandardError / ValueError / KeyError ... └── asyncio.CancelledError (Python 3.8 调整为继承自 BaseException)异步任务取消信号被吞的问题在 Agent 异步后端中如果客户端中断连接或请求超时事件循环会通过向协程注入asyncio.CancelledError来取消任务。# 错误示范随手写了 except BaseException 或裸 except:try:awaitcall_agent_step()exceptBaseExceptionase:# 吞掉了 CancelledErrorlogger.error(发生错误继续兜底)# 导致任务无法被正常取消后台协程形成僵尸死循环持续浪费 GPU/网络资源核心防范准则业务逻辑兜底捕获必须严格限定在except Exception as e:绝不捕获BaseException如果确需在底层捕获所有异常进行日志记录必须将asyncio.CancelledError重新向上抛出try:awaitbackground_job()exceptasyncio.CancelledError:logger.info(任务正常被取消执行资源清理...)raise# 必须重新抛出让事件循环感知取消exceptExceptionase:logger.exception(f业务执行失败:{e})