本文适合谁理解 Java 注解Annotation机制、想搞清楚 Python 装饰器如何工作的工程师。读完本篇你能看懂 LangChain 的tool、FastAPI 的app.get、LangSmith 的traceable背后的原理。LangChain 的工具定义代码如下tooldefsearch_web(query:str)-str:在网上搜索信息返回搜索结果摘要。returntavily_client.search(query)tool是什么为什么加了一个tool这个函数就变成了 Agent 可以调用的工具docstring 又是怎么变成工具描述的对有 Java 背景的开发者来说这种代码模式不陌生——Spring 的Service、TransactionalMyBatis 的Select到处都是注解。但 Python 的装饰器和 Java 注解是两种完全不同的东西。Java 注解是元数据由框架在运行时用反射处理。Python 装饰器是函数调用是在函数定义时就执行的代码。理解这个区别才能真正读懂 LangChain、LangSmith、FastAPI 的源码。1.1 装饰器的本质函数是一等公民装饰器的包装机制与执行顺序Python 里函数是一等公民。这句话的意思是函数和字符串、数字、列表一样可以赋值给变量可以作为参数传递可以作为返回值。defgreet(name:str)-str:returnf你好{name}# 函数可以赋值给变量say_hellogreetprint(say_hello(张三))# 你好张三# 函数可以作为参数传递defapply(func,value):returnfunc(value)print(apply(greet,李四))# 你好李四Java 里做不到这一点至少不能这么直接。Java 8 之后有 Lambda 和函数式接口但依然需要类型声明和包装。装饰器就是接收函数、返回函数的高阶函数高阶函数把其他函数当作参数或返回值的函数。defmy_decorator(func):defwrapper(*args,**kwargs):print(函数执行前)resultfunc(*args,**kwargs)print(函数执行后)returnresultreturnwrappermy_decoratordefsay_hi():print(Hi!)say_hi()# 输出# 函数执行前# Hi!# 函数执行后my_decorator只是语法糖等价于say_himy_decorator(say_hi)装饰器在函数定义时就执行不是在调用时。这和 Java 注解在运行时被反射处理是本质区别。1.2 实现一个计时装饰器用装饰器实现计时对比一下 Java 的 AOPAspect-Oriented Programming面向切面编程一种把日志、事务等横切逻辑统一管理的编程方式。Java 做法是定义一个切面类用Aspect注解再用Around拦截目标方法。代码分散在两个地方读起来需要在脑子里把切面和业务代码拼起来。Python 的装饰器直接、内联importtimeimportfunctoolsdeftimer(func):functools.wraps(func)# 保留原函数的元信息defwrapper(*args,**kwargs):starttime.perf_counter()resultfunc(*args,**kwargs)elapsedtime.perf_counter()-startprint(f{func.__name__}执行耗时{elapsed:.4f}s)returnresultreturnwrappertimerdefcall_llm(prompt:str)-str:time.sleep(0.1)# 模拟 LLM 调用returnfresponse to:{prompt}call_llm(你好)# call_llm 执行耗时0.1012sfunctools.wraps(func)这一行很重要。没有它wrapper.__name__会是wrapper而不是call_llmwrapper.__doc__函数的文档字符串即紧跟在函数定义后的那段注释字符串用于说明函数用途也会丢失。LangChain 的tool装饰器依赖函数的__doc__属性获取工具描述如果不用functools.wraps工具描述就会丢失。1.3 带参数的装饰器三层嵌套有时候装饰器本身需要接收参数。以给timer加一个prefix参数为例deftimer(prefix:str):defdecorator(func):functools.wraps(func)defwrapper(*args,**kwargs):starttime.perf_counter()resultfunc(*args,**kwargs)elapsedtime.perf_counter()-start labelf[{prefix}] ifprefixelseprint(f{label}{func.__name__}耗时{elapsed:.4f}s)returnresultreturnwrapperreturndecoratortimer(prefixLLM调用)defcall_gpt(prompt:str)-str:time.sleep(0.05)returnsome responsecall_gpt(什么是 Agent)# [LLM调用] call_gpt 耗时0.0501s三层嵌套是带参数装饰器的固定模式最外层接收装饰器参数中间层接收函数最内层是实际执行逻辑。理解这个模式的方法是从内往外读最里面的wrapper是真正干活的函数中间的decorator是标准装饰器结构最外面的timer是生产装饰器的工厂。1.4 类装饰器用__call__实现除了函数类也可以作为装饰器。用__call__方法让类的实例变成可调用对象classRetryDecorator:def__init__(self,max_retries:int3,delay:float1.0):self.max_retriesmax_retries self.delaydelaydef__call__(self,func):functools.wraps(func)defwrapper(*args,**kwargs):forattemptinrange(self.max_retries):try:returnfunc(*args,**kwargs)exceptExceptionase:ifattemptself.max_retries-1:raiseprint(f第{attempt1}次失败{e}{self.delay}s 后重试)time.sleep(self.delay)returnwrapperRetryDecorator(max_retries3,delay0.5)defunstable_api_call(url:str)-dict:# 模拟不稳定的 APIimportrandomifrandom.random()0.7:raiseConnectionError(网络超时)return{status:ok}类装饰器比函数装饰器更适合有状态的场景比如记录调用次数、管理连接池。1.5 AI 开发中的常见装饰器理解了原理再看框架里的装饰器就清晰多了。toolLangChainfromlangchain.toolsimporttooltooldefget_weather(city:str)-str:获取指定城市的天气信息。city 参数是城市名称如北京、上海。# 实际调用天气 APIreturnf{city}今天晴气温 25°CLangChain 的tool做了这几件事从函数签名提取参数 schema从 docstring 提取描述把函数包装成StructuredTool对象。Agent 看到的不是函数而是一个有 name、description、args_schema 属性的对象。traceableLangSmithfromlangsmithimporttraceabletraceable(namellm-call,run_typellm)defcall_openai(messages:list)-str:responseopenai_client.chat.completions.create(modelgpt-4o,messagesmessages)returnresponse.choices[0].message.content加了traceableLangSmith 会自动记录函数的输入、输出、耗时在 LangSmith 的 UI 里可以看到完整的调用链。这个装饰器的实现原理是在函数调用前后向 LangSmith 发送 trace 数据。cachefunctoolsfromfunctoolsimportlru_cachelru_cache(maxsize128)defget_embedding(text:str)-tuple:# LLM 调用很贵相同文本不要重复 embeddingresponseopenai_client.embeddings.create(modeltext-embedding-3-small,inputtext)returntuple(response.data[0].embedding)# list 不可哈希转成 tuplelru_cache是标准库里的缓存装饰器基于 LRULeast Recently Used最近最少使用策略——当缓存满了优先淘汰最久没被用到的结果。Embedding 调用有成本缓存能减少重复请求。1.6 函数式编程基础装饰器是高阶函数的应用函数式编程的基础知识同样值得掌握。map、filter、reduce是三个经典的高阶函数fromfunctoolsimportreducescores[85,92,78,96,61,88]# filter筛选及格分数passedlist(filter(lambdax:x60,scores))# map换算成百分比percentageslist(map(lambdax:x/100,scores))# reduce求总分totalreduce(lambdaacc,x:accx,scores)在实际代码里列表推导式比map/filter更 Pythonic可读性更好# 比 filter map 的写法更直观passed_percentages[x/100forxinscoresifx60]map/filter适合函数已经存在、不需要写 lambda 的情况importjson raw_data[{name: 张三},{name: 李四}]parsedlist(map(json.loads,raw_data))# 比列表推导式简洁1.7 完整示例实现工具注册装饰器下面是一个模仿 LangChaintool实现的简化版工具注册系统可以实际运行importinspectimportfunctoolsfromtypingimportCallable,Any# 全局工具注册表_tool_registry:dict[str,dict]{}deftool(func:Callable)-Callable:把函数注册为 Agent 工具从函数签名和 docstring 提取元信息。# 提取函数签名siginspect.signature(func)params{}forname,paraminsig.parameters.items():annotationparam.annotation params[name]{type:annotation.__name__ifannotation!inspect.Parameter.emptyelseany,required:param.defaultinspect.Parameter.empty}# 提取 docstring 作为工具描述descriptioninspect.getdoc(func)or无描述# 注册到全局工具表_tool_registry[func.__name__]{name:func.__name__,description:description,parameters:params,func:func}functools.wraps(func)defwrapper(*args,**kwargs):returnfunc(*args,**kwargs)wrapper.is_toolTruewrapper.tool_info_tool_registry[func.__name__]returnwrapperdefget_tools()-list[dict]:获取所有已注册的工具列表。return[{k:vfork,vininfo.items()ifk!func}forinfoin_tool_registry.values()]# 使用示例tooldefsearch_web(query:str)-str:在互联网上搜索信息。query 是搜索关键词。returnf搜索结果关于 {query} 的信息...tooldefcalculate(expression:str)-float:计算数学表达式。expression 是合法的 Python 数学表达式如 2 3 * 4。returneval(expression)tooldefget_current_time()-str:获取当前系统时间。fromdatetimeimportdatetimereturndatetime.now().strftime(%Y-%m-%d %H:%M:%S)if__name____main__:importjson# 查看注册的工具toolsget_tools()print(已注册的工具)print(json.dumps(tools,ensure_asciiFalse,indent2))# 直接调用工具print(\n直接调用)print(search_web(Python 装饰器))print(calculate(2 ** 10))print(get_current_time())# 模拟 Agent 根据工具名调用print(\nAgent 调用)tool_namecalculatetool_args{expression:100 / 4 5}result_tool_registry[tool_name][func](**tool_args)print(f调用{tool_name}结果{result})运行这段代码工具的 name、description、parameters 会被自动提取出来结构和 LangChain 工具的 schema 非常接近。1.8 装饰器在 LangChain 中的实际应用理解了装饰器原理看 LangChain 的实际代码会更清晰。这里用三个真实场景演示1.8.1 场景1tool 定义 LangChain 工具fromlangchain_core.toolsimporttoolfrompydanticimportBaseModel,Field# 方式一直接用 tool简单工具tooldefget_weather(city:str)-str:获取指定城市的当前天气。city 是中文城市名如北京。# 这里 tool 做了什么# 1. 从函数名提取工具名称get_weather# 2. 从 docstring 提取工具描述# 3. 从类型注解生成参数 schema{city: {type: string}}# 4. 把函数包装成 StructuredTool 对象Agent 可以调用returnf{city}今天晴气温 25°C# 查看工具的元信息Agent 调用前会看这些print(get_weather.name)# get_weatherprint(get_weather.description)# 获取指定城市的当前天气。...print(get_weather.args)# {city: {title: City, type: string}}# 方式二用 pydantic 定义复杂参数推荐参数描述更精确classSearchInput(BaseModel):query:strField(description搜索关键词尽量简洁)max_results:intField(default5,ge1,le20,description返回结果数量)tool(args_schemaSearchInput)defsearch_web(query:str,max_results:int5)-list[str]:搜索互联网获取最新信息。当需要查询实时数据时使用。# Field 的 description 会传给 LLM帮助它理解如何填写参数return[f搜索结果{i}:{query}相关内容foriinrange(max_results)]1.8.2 场景2traceable 追踪 LLM 调用链路fromlangsmithimporttraceablefromopenaiimportOpenAI clientOpenAI()traceable(namellm-call,run_typellm)defcall_openai(messages:list[dict],model:strgpt-4o)-str: traceable 做了什么 - 记录函数调用的输入messages, model - 记录函数调用的输出返回的文字 - 记录耗时 - 上报到 LangSmith在 UI 里可以看到完整链路 responseclient.chat.completions.create(modelmodel,messagesmessages)returnresponse.choices[0].message.contenttraceable(namerag-pipeline)defrag_pipeline(question:str)-str:整个 RAG 流程也可以追踪LangSmith 会展示嵌套调用树docsretrieve_documents(question)# 内部调用也可以加 traceablecontextformat_context(docs)answercall_openai([# 这里会嵌套在 rag-pipeline 下{role:system,content:f参考资料{context}},{role:user,content:question}])returnanswer1.8.3 场景3app.post FastAPI 路由注册fromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModel appFastAPI()classChatRequest(BaseModel):message:strsession_id:str|NoneNoneclassChatResponse(BaseModel):reply:strtokens_used:intapp.post(/chat,response_modelChatResponse)asyncdefchat_endpoint(request:ChatRequest)-ChatResponse: app.post(/chat) 做了什么 - 注册 POST /chat 路由 - 从 ChatRequest 的类型注解自动生成请求 body 的 JSON Schema - 从 ChatResponse 的类型注解自动生成响应的 JSON Schema - 在 /docs 页面自动生成交互式 API 文档 try:replycall_openai([{role:user,content:request.message}])returnChatResponse(replyreply,tokens_used100)exceptExceptionase:raiseHTTPException(status_code500,detailstr(e))1.9 总结装饰器来自做了什么Java 等价toolLangChain提取元信息注册为 Agent 工具Service XML 配置traceableLangSmith记录输入输出上报链路追踪Around 切面app.postFastAPI注册路由生成 API 文档PostMappinglru_cache标准库缓存函数返回值按参数去重Spring Cacheableretrytenacity失败自动重试支持指数退避Spring Retry装饰器的本质是接收函数、返回函数的高阶函数加上语法糖。掌握这一点再去看 LangChain 的tool、LangSmith 的traceable、FastAPI 的app.get就能看清楚每个装饰器背后做了什么这里用了functools.wraps那里用了三层嵌套这里在函数定义时做了注册。Python 框架大量使用装饰器不是为了炫技而是因为这种模式让调用侧极其简洁——一行tool就完成了工具注册而不需要像 Java 那样写一堆配置类。理解装饰器是读懂 AI 框架源码的前提。