Day11：从零开始写一个Agent项目（3）

开头大家好这里是惬鹤频道那么随着agent项目的学习进入了第三天这次要给大家分享的是关于agent部分代码的编写今天的代码不算很难很快就可以理解那让我们开始吧如果你觉得这篇文章有用请点点赞谢谢可以的话关注我吧我会努力更新的总会有的。。。项目结构图代码文件rag_service.py 文件名rag_service.py 描述用于总结服务 # 依赖文件导入fromrag.vector_storeimportVectorStoreServicefromutils.prompt_loaderimportload_rag_promptsfrommodel.factoryimportchat_model# 依赖导入fromlangchain_core.promptsimportPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.documentsimportDocument# 打印提示词defprint_prompt(prompt):print(*20)print(prompt.to_string())print(*20)returnprompt# 定义RAG总结服务类classRagSummarizeService(object):def__init__(self):self.vector_storeVectorStoreService()self.retrieverself.vector_store.get_retriever()self.prompt_textload_rag_prompts()self.prompt_templatePromptTemplate.from_template(self.prompt_text)self.modelchat_model self.chainself.__init__chain()# 初始化链def__init__chain(self):chainself.prompt_template|print_prompt|self.model|StrOutputParser()returnchain# 得到检索问题相关资料的结果相关参数如k已经设置在配置文件中defretriever_docs(self,query:str)-list[Document]:returnself.retriever.invoke(query)# 将用户问题转换为可插入提示词的一部分并返回调用结果defrag_summarize(self,query:str)-str:# 拿到Document类型的列表context_docsself.retriever_docs(query)counter0contextfordocincontext_docs:counter1contextf[参考资料{counter}] 资料内容{doc.page_content}| 参考源数据{doc.metadata}\n# 返回链调用的结果returnself.chain.invoke({input:query,context:context,})# 测试if__name____main__:ragRagSummarizeService()print(rag.rag_summarize(小户型适合哪种扫地机器人))这是一个 RAG 总结服务类它通过向量检索找到与问题相关的文档片段将它们拼接成上下文再交给大语言模型生成最终回答。类内部封装了检索器、提示词模板和 LCEL 链对外提供 rag_summarize(query) 方法即可完成从问题到答案的完整流程。这段代码的结构分为以下几个层次模块导入部分从项目内部模块导入rag.vector_store.VectorStoreService向量库服务负责文档检索utils.prompt_loader.load_rag_prompts加载提示词模板model.factory.chat_model聊天模型实例从 LangChain 导入PromptTemplate提示词模板StrOutputParser输出解析器将模型输出转为字符串Document文档类型注解辅助函数print_prompt(prompt)打印完整的提示词内容用于调试并原样返回 prompt核心类 RagSummarizeService内部有如下几个方法初始化init创建向量库服务实例 self.vector_store获取检索器 self.retriever用于根据问题检索相关文档加载提示词文本 self.prompt_text创建提示词模板 self.prompt_template引用全局聊天模型 self.model调用私有方法 __init__chain() 初始化 LCEL 链私有方法 __init__chain构造链prompt_template | print_prompt | model | StrOutputParser()返回构建好的链支持 invoke检索方法 retriever_docs(query)调用检索器的 invoke 方法返回相关文档列表list[Document]核心方法 rag_summarize(query)调用 retriever_docs(query) 获取相关文档列表遍历文档格式化为带编号和元数据的上下文字符串 context调用链的 invoke传入 {“input”: query, “context”: context}返回最终的回答字符串整体的结构如下RagSummarizeService├──init│ ├── 初始化 VectorStoreService│ ├── 获取 retriever│ ├── 加载 prompt 文本│ ├── 创建 PromptTemplate│ ├── 引用 chat_model│ └── 调用 __init__chain 创建 chain├── retriever_docs(query) → list[Document]├── rag_summarize(query) → str│ ├── 调用 retriever_docs 获取相关文档│ ├── 将文档格式化为 context 字符串│ └── chain.invoke({“input”: query, “context”: context})└── (内置 print_prompt 辅助函数)代码文件agent_tools.py 文件名agent_tools 描述智能体需要使用的各种方法 importos# 依赖导入fromlangchain_core.toolsimporttoolimportrandom# 依赖文件导入fromrag.rag_serviceimportRagSummarizeServicefromutils.config_handlerimportagent_conffromutils.path_toolimportget_abs_pathfromutils.logger_handlerimportlogger ragRagSummarizeService()user_ids[1001,1002,1003,1004,1005,1006,1007,1008,1009,1010,]month_arr[2025-01,2025-02,2025-03,2025-04,2025-05,2025-06,2025-07,2025-08,2025-09,2025-10,2025-11,2025-12,]external_data{}tool(description从向量存储中检索参考资料)defrag_summarize(query:str)-str:returnrag.rag_summarize(query)tool(description获取城市的天气这里使用了固定值。)defget_weather(city:str)-str:returnf城市{city}天气为晴天气温26摄氏度空气湿度50%南风1级AQI21最近六小时降雨概率极低。tool(description获取用户所在城市的名称)defget_user_location()-str:returnrandom.choice([北京,厦门,莆田])tool(description获取用户的ID)defget_user_id()-str:returnrandom.choice(user_ids)tool(description获取当前月份)defget_current_month()-str:returnrandom.choice(month_arr)defgenerate_external_data():ifnotexternal_data:external_data_pathget_abs_path(agent_conf[external_data_path])ifnotos.path.exists(external_data_path):raiseFileNotFoundError(f外部数据文件{external_data_path}不存在)withopen(external_data_path,r,encodingutf-8)asf:forlineinf.readlines()[1:]:arr:list[arr]line.strip().split(,)user_id:strarr[0].replace(,)feature:strarr[1].replace(,)efficiency:strarr[2].replace(,)consumables:strarr[3].replace(,)comparison:strarr[4].replace(,)time:strarr[5].replace(,)ifuser_idnotinexternal_data:external_data[user_id]{}external_data[user_id][time]{特征:feature,效率:efficiency,耗材:consumables,对比:comparison,}tool(description从外部系统的获取用户的使用记录)deffetch_external_data(user_id:str,month:str)-str:generate_external_data()try:returnexternal_data[user_id][month]exceptKeyError:logger.warning(f[方法fetch_external_data]未能检索到用户{user_id}在{month}的使用记录)return# 测试if__name____main__:print(fetch_external_data(user_id1001,month2025-01))代码中的注释比较少下面进行介绍这段代码写了一些可供Agent调用的方法有如下几种rag_summarize 从向量数据库检索知识get_weather 查询城市天气get_user_location 获取用户城市get_user_id 获取用户IDget_current_month 获取当前月份fetch_external_data 获取用户某月使用记录还包含一个辅助函数 generate_external_data用来读取外部CSV文件将数据存入 external_data 字典。基本结构如下agent_tools.py├── 导入依赖├── 全局实例化rag RagSummarizeService()├── 静态数据user_ids, month_arr, external_data├── 工具定义tool 装饰的6个函数│ ├── rag_summarize(query) → RAG检索│ ├── get_weather(city) → 模拟天气│ ├── get_user_location() → 模拟用户位置│ ├── get_user_id() → 模拟用户ID│ ├── get_current_month() → 模拟当前月份│ └── fetch_external_data(user_id, month) → 从CSV查使用记录├── 辅助函数generate_external_data() → 加载CSV到external_data└── 测试入口这个文件对Agent来说很重要它让Agent具备执行具体操作的能力。同时在实现功能的过程中也借助了utils文件夹内的工具文件和RAG检索文件。结尾这期要分享的代码就是这些同时这次项目编写也差不多要接近尾声了下一次大概就是最后一期。老样子在完结后我会把所有代码放到GitHub上需要研究的可以下载。感谢大家的支持我们下期再见