一份招聘要求让我重新认识了什么是“AI 应用开发”一、一份招聘要求引发的思考前几天刷招聘网站看到一份 Python 后端 LLM 岗位的 JD看完之后我沉默了。岗位要求分四块基础能力、AI/LLM能力、工程能力、加分项。乍一看好像也没什么特别的——FastAPI、Docker、Git、PostgreSQL这些不都是后端基本功吗但再看第二块画风突变理解 LLM 基本概念Token、Temperature、上下文窗口、Embedding必须有 LangChain 或 LlamaIndex 的实际项目使用经验熟悉 Prompt Engineering会使用输出格式约束了解 RAG 基本流程用过至少一种向量数据库第三块更狠Python 异步编程处理并发请求API 异常处理策略重试、降级、熔断日志和监控意识加分项直接指向生产级别的 AI Agent 项目经验——上线规模、日均调用量、性能指标、踩过的坑。这哪是招后端这分明是在招一个能把大模型应用真正落地到生产环境的人。而其中最扎眼的一条是必须有 LangChain 的实际项目使用经验。说实话我之前对 LangChain 的理解还停留在“调个 API、拼个 Prompt”的层面。但这份 JD 让我意识到——如果只会调 API连简历关都过不去。于是我连夜打开了 LangChain 的源码。二、LangChain 到底是什么先别急着写代码很多人第一次接触 LangChain 的时候会有一个误区觉得 LangChain 是一个大模型。其实不是。LangChain 不是 ChatGPT不是 DeepSeek不是 Qwen也不是 Claude。它更像是一个大模型应用开发框架。用一句大白话解释大模型负责“生成答案”LangChain 负责“把大模型和 Prompt、知识库、工具、数据库、接口、业务流程串起来”。如果把大模型比作一个聪明的大脑LangChain 更像是神经系统、手脚和流程控制器。大脑可以思考但它自己不会查数据库不会调用接口不会读取企业文档不会自动决定走哪个业务流程。LangChain 要解决的就是这些工程化问题。它的核心定位不是“替代大模型”而是帮助开发者更方便地构建大模型应用。它主要解决这些问题怎么管理 Prompt怎么统一调用不同模型怎么接入知识库怎么做 RAG怎么让模型调用工具怎么让多个步骤串起来怎么让输出结果结构化怎么做日志和链路追踪LangChain 最重要的思想就是把复杂流程拆成组件然后像搭积木一样组合起来。而把组件串起来的核心骨架就是Chain。三、源码深挖Chain——流程编排的核心骨架打开 LangChain 的源码最先映入眼帘的是langchain/chains/base.py。Chain 的官方定义是创建对组件进行结构化序列调用的抽象基类。说白了Chain 就是把多个步骤按逻辑串联成可执行的工作流。无论是简单的“提问-回答”流程还是复杂的“检索-思考-工具调用”pipeline都依赖 Chain 基类提供的基础能力。3.1 Chain 解决了什么问题看源码会发现Chain 基类解决了三个关键问题第一统一接口。所有 Chain 遵循相同的调用方式invoke、ainvoke降低使用和扩展成本。第二流程标准化。封装“输入处理→核心执行→输出处理”的通用流程避免重复开发。第三能力集成。内置 Memory记忆、Callbacks回调等通用能力让所有子类自动具备状态管理、可观察性等特性。3.2 Chain 的核心接口源码级解读Chain 作为抽象基类ABC定义了子类必须实现的几个核心接口。1input_keys 与 output_keys输入输出的“规格说明”propertyabstractmethoddefinput_keys(self)-List[str]:Chain 运行所需的输入参数名称propertyabstractmethoddefoutput_keys(self)-List[str]:Chain 输出的结果参数名称这两个抽象属性定义了 Chain 的“输入输出规格”input_keys指定需要哪些输入如问答 Chain 可能需要[question, context]output_keys指定输出什么如问答 Chain 可能输出[answer]2_call 与 _acall核心执行逻辑abstractmethoddef_call(self,inputs:Dict[str,Any],run_manager:Optional[CallbackManagerForChainRun]None,)-Dict[str,Any]:同步执行 Chain 核心逻辑asyncdef_acall(self,inputs:Dict[str,Any],run_manager:Optional[AsyncCallbackManagerForChainRun]None,)-Dict[str,Any]:异步执行 Chain 核心逻辑默认基于同步方法实现_call是所有 Chain 的“心脏”子类必须实现具体的业务逻辑输入inputs符合input_keys规格的参数字典输出符合output_keys规格的结果字典run_manager用于触发回调事件执行开始、结束、错误异步支持方面_acall默认通过run_in_executor调用同步_call子类可重写以实现原生异步逻辑。3.3 常见的 Chain 类型LangChain 提供了多种 Chain 实现Chain 类型适用场景LLMChain单模型调用链SequentialChain顺序执行多个 ChainSimpleSequentialChain线性任务流RouterChain条件分支/规则路由TransformationChain数据转换3.4 实战自定义一个 Chain理解了 Chain 的源码我们就可以自己实现一个fromtypingimportList,Dict,Any,Optionalfromlangchain.chains.baseimportChainfromlangchain.callbacks.managerimportCallbackManagerForChainRunclassAddChain(Chain):一个简单的加法 Chainpropertydefinput_keys(self)-List[str]:return[a,b]propertydefoutput_keys(self)-List[str]:return[sum]def_call(self,inputs:Dict[str,Any],run_manager:Optional[CallbackManagerForChainRun]None,)-Dict[str,Any]:ainputs[a]binputs[b]return{sum:ab}# 使用chainAddChain()resultchain.invoke({a:3,b:5})print(result)# {sum: 8}这个例子虽然简单但它揭示了 Chain 的本质定义输入→执行业务逻辑→返回输出。所有复杂的 RAG Chain、Agent Chain都是在这个骨架之上生长出来的。四、源码深挖Retriever——RAG 的检索心脏岗位要求里明确提到了RAG 基本流程和向量数据库。而 RAG 的核心就是 Retriever检索器。4.1 Retriever 是什么检索器Retriever是 LangChain 中的一个核心组件它提供了一个通用接口用于根据文本查询返回相关的文档。关键点检索器比向量存储更加通用。它不需要能够存储文档只需要能够返回或检索文档。向量存储可以用作检索器的后端但也存在其他类型的检索器。4.2 为什么需要 Retriever语言模型存在三大局限知识过时LLM 的知识截止于训练时间上下文不足无法访问特定领域的专有数据幻觉问题在缺乏事实依据时可能生成不准确的内容Retriever 正是为了解决这些问题而存在的。RAG 的流程就是用户查询 → 检索器 → 相关文档 → LLM → 生成答案4.3 Retriever 的核心接口源码级查看langchain_core/retrievers.py的源码classBaseRetriever(Serializable,ABC):文档检索系统的抽象基类abstractmethoddef_get_relevant_documents(self,query:str,*,run_manager:Optional[CallbackManagerForRetrieverRun]None)-List[Document]:子类必须实现的核心检索逻辑asyncdef_aget_relevant_documents(self,query:str,*,run_manager:Optional[AsyncCallbackManagerForRetrieverRun]None)-List[Document]:异步检索可选实现子类必须实现_get_relevant_documents方法。异步版本_aget_relevant_documents是可选的。4.4 实现自定义 RetrieverfromtypingimportListfromlangchain_core.retrieversimportBaseRetrieverfromlangchain_core.documentsimportDocumentclassSimpleRetriever(BaseRetriever):简单检索器示例docs:List[Document]k:int5def_get_relevant_documents(self,query:str,*,run_managerNone)-List[Document]:返回文档列表中的前 k 个文档returnself.docs[:self.k]asyncdef_aget_relevant_documents(self,query:str,*,run_managerNone)-List[Document]:异步实现returnself.docs[:self.k]4.5 从向量存储创建 Retriever在实际项目中我们通常从向量存储创建 Retrieverfromlangchain_community.vectorstoresimportChromafromlangchain_openaiimportOpenAIEmbeddings# 创建向量存储vectorstoreChroma.from_documents(documentssplits,embeddingOpenAIEmbeddings(),persist_directory./chroma_db)# 转换为检索器retrievervectorstore.as_retriever(search_typesimilarity,# 搜索类型search_kwargs{k:5}# 返回 top-k)检索器还可以轻松转换为工具供 Agent 使用fromlangchain_core.toolsimportcreate_retriever_tool retriever_toolcreate_retriever_tool(retriever,search_knowledge_base,用于搜索知识库的检索工具)五、源码深挖Tool——让 LLM 拥有“手脚”岗位要求里提到了Tool是 LangChain 的核心组件之一。Tool 是让 LLM 能够调用外部函数、查询数据库、执行操作的关键机制。5.1 Tool 的本质很多人第一次看 LangChain Tool 调用时会误以为 LLM 直接调用了 Python 函数。实际上不是。LLM 不会进入你的 Python 进程也不会真的执行search_weather()、calculator()这些函数。真正发生的是Python 函数 → tool 包装成 BaseTool → create_agent() 把工具 schema 绑定给模型 → 模型返回 AIMessage.tool_calls → LangGraph 条件边路由到 ToolNode → ToolNode 在本地执行 Python 函数 → 执行结果变成 ToolMessage → ToolMessage 回到 messages模型继续推理工具调用的本质是三层协作层级负责什么Tool 定义层把函数变成带 name、description、args schema 的工具对象Agent 编排层绑定工具 schema、构建图节点、决定是否进入工具节点Tool 执行层解析 tool call、执行本地工具、生成 ToolMessage5.2 tool 装饰器的源码解析先看一个最简单的例子fromlangchain.toolsimporttooltooldefadd(a:int,b:int)-int:相加returnabtool的源码大致如下deftool(_funcNone,*,args_schemaNone,return_directFalse,**kwargs):defdecorator(func):returnTool.from_function(func,args_schemaargs_schema,return_directreturn_direct,**kwargs,)if_funcisNone:returndecoratorelse:returndecorator(_func)tool支持无参数和有参数两种装饰用法真正的工作由Tool.from_function完成。5.3 Tool.from_function 源码解读classTool(BaseTool):classmethoddeffrom_function(cls,func:Callable,name:Optional[str]None,description:Optional[str]None,args_schema:Optional[Type[BaseModel]]None,return_direct:boolFalse,**kwargs,):# 自动获取函数名字和文档说明tool_namenameorfunc.__name__ tool_descriptiondescriptionorfunc.__doc__or# 关键点1自动生成参数 schemaifargs_schemaisNone:args_schemacreate_schema_from_function(func)# 关键点2封装函数执行逻辑returncls(nametool_name,funcfunc,args_schemaargs_schema,descriptiontool_description,return_directreturn_direct,**kwargs,)两个关键点参数定义自动生成默认会从函数的参数签名和类型注解动态构造 Pydantic schema这决定了 Agent 能不能理解参数类型。函数封装原始函数直接挂载到 Tool 对象的func属性上后续通过 Tool 对象统一调度。六、实战从零搭建一个 RAG Agent 智能问答系统理论看完了我们来写一个能跑起来的项目。这个项目会同时用到 Chain、Retriever、Tool 三大组件。6.1 环境准备pipinstalllangchain langchain-community langchain-openai chromadb pypdf6.2 完整代码importosfromtypingimportListfromlangchain_community.document_loadersimportPyPDFLoaderfromlangchain_text_splittersimportRecursiveCharacterTextSplitterfromlangchain_openaiimportOpenAIEmbeddings,ChatOpenAIfromlangchain_community.vectorstoresimportChromafromlangchain.chainsimportRetrievalQAfromlangchain_core.toolsimporttoolfromlangchain.agentsimportcreate_agent# 1. 加载和分割文档 defload_and_split_documents(file_path:str):加载 PDF 并分割成 chunksloaderPyPDFLoader(file_path)documentsloader.load()text_splitterRecursiveCharacterTextSplitter(chunk_size500,chunk_overlap100,# 重叠防止上下文断裂)returntext_splitter.split_documents(documents)# 2. 创建向量存储和检索器 defcreate_retriever(splits:List,persist_dir:str./chroma_db):创建向量存储并返回检索器vectorstoreChroma.from_documents(documentssplits,embeddingOpenAIEmbeddings(),persist_directorypersist_dir)vectorstore.persist()returnvectorstore.as_retriever(search_typesimilarity,search_kwargs{k:5})# 3. 自定义工具 tooldefget_current_time()-str:获取当前时间fromdatetimeimportdatetimereturndatetime.now().strftime(%Y-%m-%d %H:%M:%S)tooldefcalculate(expression:str)-str:计算数学表达式例如 2 3 * 4try:resulteval(expression)returnf计算结果:{result}exceptExceptionase:returnf计算错误:{str(e)}# 4. 创建 RAG Chain defcreate_qa_chain(retriever,llm):创建检索增强生成的问答链returnRetrievalQA.from_chain_type(llmllm,chain_typestuff,retrieverretriever,return_source_documentsTrue,verboseTrue)# 5. 创建 Agent含工具和RAG defcreate_agent_with_tools(llm,retriever_tool):创建带有工具调用能力的 Agenttools[retriever_tool,get_current_time,calculate]agentcreate_agent(modelllm,toolstools,system_prompt你是一个智能助手可以 1. 使用 search_knowledge_base 工具搜索知识库中的信息 2. 使用 get_current_time 获取当前时间 3. 使用 calculate 进行数学计算 请根据用户问题选择合适的工具。)returnagent# 6. 主程序 defmain():# 初始化 LLMllmChatOpenAI(modelgpt-4o-mini,temperature0.3,# 降低随机性提高准确性)# 加载文档假设有一份 company_policy.pdftry:splitsload_and_split_documents(company_policy.pdf)print(f文档已分割为{len(splits)}个 chunks)exceptFileNotFoundError:print(未找到文档使用示例文档...)# 创建示例文档fromlangchain_core.documentsimportDocument splits[Document(page_content公司年假政策入职满1年享有5天年假满3年享有10天。),Document(page_content公司调薪政策每年4月进行年度调薪幅度为3%-10%。),Document(page_content公司福利五险一金、补充医疗保险、年度体检。)]# 创建检索器retrievercreate_retriever(splits)print(✅ 检索器创建成功)# 创建检索器工具fromlangchain_core.toolsimportcreate_retriever_tool retriever_toolcreate_retriever_tool(retriever,search_knowledge_base,搜索公司政策知识库用于回答关于公司政策、福利、制度的问题)# 创建 Agentagentcreate_agent_with_tools(llm,retriever_tool)print(✅ Agent 创建成功)# 测试 test_questions[公司的年假政策是什么,现在几点了,计算 123 * 456 789,]forquestionintest_questions:print(f\n{*50})print(f用户:{question})print(f{*50})resultagent.invoke({messages:[{role:user,content:question}]})print(f助手:{result[messages][-1].content})if__name____main__:main()6.3 代码解读这个项目展示了 LangChain 三大核心组件的协同工作ChainRetrievalQA封装了“检索→生成”的完整流程用户只需要输入问题Chain 自动完成检索和生成。Retriever从 Chroma 向量数据库中检索相关文档作为 LLM 生成答案的知识来源。Tool将get_current_time和calculate两个普通函数包装成工具让 Agent 可以调用。Agent 的决策流程是model 节点调用 LLM根据对话历史和工具描述决定是直接回答还是调用工具tools 节点执行工具调用返回结果条件判断有 tool_calls → 跳转到 tools无 tool_calls → 结束七、生产环境踩过的那些坑源码看完了代码也写了但真正把 LangChain 应用部署到生产环境还有一堆坑等着你。7.1 版本锁定是血泪教训“LangChain in production is tricky - dependency management and error handling will make or break you. I’ve deployed three different systems this year and learned the hard way that you MUST pin your versions.”LangChain 的版本迭代非常快不同版本之间的 API 可能完全不兼容。必须锁定版本否则今天能跑的代码明天就可能报错。7.2 API Key 安全管理LangChain 常需访问 API 密钥如OPENAI_API_KEY直接硬编码至镜像存在安全风险。推荐通过 Docker Secrets 或环境变量注入敏感信息。# Dockerfile 中不要写死 API Key # 通过环境变量注入 ENV OPENAI_API_KEY7.3 异步处理并发请求岗位要求里明确提到了Python 异步编程。在 Web 服务中必须使用ainvoke而不是invokefromfastapiimportFastAPIfrompydanticimportBaseModel appFastAPI()classQueryRequest(BaseModel):question:strapp.post(/ask)asyncdefask(request:QueryRequest):# 使用异步方法处理并发请求resultawaitagent.ainvoke({messages:[{role:user,content:request.question}]})return{answer:result[messages][-1].content}7.4 日志和链路追踪岗位要求里提到“能合理打点记录关键调用链路”importloggingfromlangchain.chains.baseimportChainclassDebugChain(Chain):def__init__(self,chain:Chain):self.chainchain self.loggerlogging.getLogger(__name__)def_call(self,inputs:dict):self.logger.info(fInput received:{inputs})resultself.chain._call(inputs)self.logger.info(fOutput generated:{result})returnresult生产环境建议配合LangSmith进行全链路追踪和调试。八、总结看完这份 JD我最大的感触是AI 应用开发已经从一个“调 API 就能搞定”的玩具阶段进入了需要扎实工程能力的生产阶段。LangChain 的源码其实没有那么可怕。它的核心思想很清晰Chain负责流程编排——把多个步骤串起来Retriever负责知识检索——让 LLM 能“翻书”Tool负责能力扩展——让 LLM 能“动手”Agent负责自主决策——让 LLM 能“思考”该用什么工具理解了这四个组件你就理解了 LangChain 的 80%。剩下的 20%是在生产环境中踩坑、优化、迭代积累的经验。而这恰恰是这份 JD 最看重的东西。如果你也在准备 AI 后端的面试不妨打开 LangChain 的源码看一看。不是为了背源码而是为了理解——当一个 AI 应用真正跑起来的时候每一行代码都在做什么。参考资料LangChain GitHub 源码 -langchain/chains/base.pyLangChain GitHub 源码 -langchain_core/retrievers.pyLangChain 源码剖析二Chain 基类源码剖析LangChain 源码分析十一RAG 检索langchain tools 源码解析以及扩展LangChain 源码解析Function Call 是如何被执行的工具调用全链路源码追踪LangChain 的原理以及源码解读基于 LangChain 构建高效 RAG 应用LangChain 2025 报告Agent 落地的残酷真相与出路