agent
参考: wow-agent
什么是 AI Agent ?
LLMs强大的语言理解和生成能力，为构建Agents奠定了基础。然而，LLMs的局限性也显而易见：
- 知识局限性：LLMs的知识仅限于训练数据，无法获取实时信息和外部知识库。
- 行动局限性：LLMs无法与外界交互，无法执行实际操作。 为了克服这些局限性，谷歌的研究人员在《New whitepaper Agents》详细论述了“Agent”的概念，将LLMs与工具和编排层相结合，赋予其自主行动的能力。
AI Agent 是目前大模型应用领域的主流研究方向，其核心价值在于借助先进AI技术，打造出更加智能、更加自主的系统。
Agent不仅在处理复杂自然语言任务方面游刃有余，更在图像生成、数据分析、教育指导等多个前沿领域展现出卓越的应用潜力。
关于 AI Agent 框架的定义，目前最广为流传的版本出自 前OpenAI应用研究主管翁丽莲(Lilian Weng)的一篇blog: LLM Powered Autonomous Agents
在博客里，Lilian 将 Agents 定义为 LLM + memory + planning skills + tool use，即大语言模型、记忆、任务规划、工具使用的集合。
其中，LLM是Agent的大脑，属于“中枢”模型，要求有以下3种能力：
- planning skills：对问题进行拆解得到解决路径，既进行任务规划
- tool use：评估自己所需的工具，进行工具选择，并生成调用工具请求
- memory：短期记忆包括工具的返回值，已经完成的推理路径；长期记忆包括可访问的外部长期存储，例如知识库最早由不仅具备了感知、记忆和决策的全面能力，更能在复杂多变的环境中自主执行任务，并通过与外部工具的紧密协作，显著提升任务处理的效率与准确性。
更多相关知识可见大模型关键技术与应用 ，这篇文章介绍了自ChatGPT发布以来，大模型关键技术和应用的主要进展。
Agents 的核心组件
Agents的核心组件 一个完整的Agent主要由三个核心组件构成：
1. 模型 (Model)
-
角色：作为Agent的“大脑”，负责理解用户输入，进行推理和规划，并选择合适的工具进行执行。
-
类型：常用的模型包括ReAct、Chain-of-Thought、Tree-of-Thought等，它们提供不同的推理框架，帮助Agent进行多轮交互和决策。
-
重要性：模型是Agent的核心，其推理能力决定了Agent的行动效率和准确性。
2. 工具 (Tools)
-
角色：作为Agent与外界交互的“桥梁”，允许Agent访问外部数据和服务，执行各种任务。
-
类型：工具可以是各种API，例如数据库查询、搜索引擎、代码执行器、邮件发送器等。
-
重要性：工具扩展了Agent的能力，使其能够执行更复杂的任务。
3. 编排层 (Orchestration Layer)
-
角色：负责管理Agent的内部状态，协调模型和工具的使用，并根据目标指导Agent的行动。
-
类型：编排层可以使用各种推理框架，例如ReAct、Chain-of-Thought等，帮助Agent进行规划和决策。
-
重要性：编排层是Agent的“指挥中心”，负责协调各个组件，确保Agent的行动符合目标。
Agent 的运作机制
从输入到输出，Agent的运作过程可以概括为以下几个步骤：
1.接收输入：Agent接收用户的指令或问题。
2.理解输入：模型理解用户的意图，并提取关键信息。
3.推理规划：模型根据用户输入和当前状态，进行推理和规划，确定下一步行动。
4.选择工具：模型根据目标选择合适的工具。
5.执行行动：Agent使用工具执行行动，例如查询数据库、发送邮件等。
6.获取结果：Agent获取工具执行的结果。
7.输出结果：Agent将结果输出给用户，或进行下一步行动。
Agents的优势
与传统的LLMs相比，Agents具有以下优势：
-
知识扩展：通过工具，Agent可以访问实时信息和外部知识库，突破训练数据的限制，提供更准确和可靠的信息。
-
自主行动：Agent可以根据目标进行自主决策和行动，无需人工干预，提高效率和灵活性。
-
多轮交互：Agent可以管理对话历史和上下文，进行多轮交互，提供更自然和流畅的用户体验。
-
可扩展性：Agent可以通过添加新的工具和模型，扩展其功能和应用范围。
Agents的应用
Agents的应用范围非常广泛，例如：
-
智能客服：Agent可以自动回答用户问题，处理订单，解决客户问题，提高客户满意度。
-
个性化推荐：Agent可以根据用户的兴趣和行为，推荐商品、内容、服务等，提升用户体验。
-
虚拟助手：Agent可以帮助用户管理日程、预订行程、发送邮件等，提高工作效率。
-
代码生成：Agent可以根据用户的需求，自动生成代码，提高开发效率。
-
智能创作：Agent可以根据用户的需求，创作诗歌、小说、剧本等，激发创作灵感。
-
知识图谱构建：Agent可以从文本中提取知识，构建知识图谱，用于知识管理和推理。
Agents的开发工具
为了方便开发Agents，Google提供了多种工具和平台，例如：
-
LangChain：一个开源库，可以帮助开发者构建和部署Agents。LangChain提供了一套API，方便开发者将LLMs与工具和编排层结合，构建功能强大的Agents。
-
LangGraph：一个开源库，可以帮助开发者构建和可视化Agents。LangGraph提供了一套图形化界面，方便开发者设计和测试Agents。
-
Vertex AI：一个云平台，提供各种AI工具和服务，例如Vertex Agent Builder、Vertex Extensions、Vertex Function Calling等，可以帮助开发者快速构建和部署Agents。Vertex AI提供了强大的基础设施和工具，方便开发者进行Agent开发、测试、部署和管理。
简易 Agent 实现
在了解完相关概念后，我们正式开始学习 agent，本次使用wow-agent作为示例讲解agent实现。
通过观察市场上比较流行的开源多智能体框架，例如 metagpt、crewai、camel-ai、autogen，我们会发现这些框架安装起来有很多依赖，但是如果我们只是想要实现一个简易的功能，就不需要这么多的依赖环境，所以，wow-agent 应运而生。
wow-agent致力于在代码行数和依赖库数量之间取得均衡的最小值，用最划算的方式在本地搭建AI Agent，嵌入到生产工作环节中。
环境准备
先准备需要的python库： OpenAI、Python-dotenv
pip install openai python-dotenv
-
OpenAI库是OpenAI官方提供的Python SDK，旨在帮助开发者轻松调用OpenAI的API，实现各种AI功能。通过这个库，开发者可以快速集成GPT、DALL·E等先进模型，构建智能应用。
-
Python-dotenv是一个用于管理环境变量的Python库。它可以从.env文件中读取键值对，并将它们设置为环境变量。这个库特别适用于管理敏感信息，如API密钥、数据库连接字符串等。
为了调用大模型，我们需要准备：
- api_key，这个需要到各家的开放平台上去申请。
- base_url，这个需要到各家的开放平台上去拷贝。
- 所用模型名称。
在项目的根目录新建一个txt文件，把文件名改成.env 并在文件中写入自己的api-key
DeepSeek_API_KEY=sk-**********************
把DeepSeek_API_KEY写到.env文件的原因是为了保密，同时可以方便地在不同的代码中读取。
在大模型厂商官网注册后，能够获取到一个api_key，一般官网文档会提供api结构的调用方式
对于本地ollama的LLM，ollama默认没有api_key，需要修改请求格式，或者使用one-api整合使模型能够使用openai的调用方式，方法请自行百度，在此不做赘述。
本次演示使用第三方平台(硅基流动)中的 deepseek-ai/DeepSeek-V3 模型，调用参考官方api文档
现在，我们先把所需的环境写好
import os from dotenv import load_dotenv # 加载环境变量 load_dotenv() # 从环境变量中读取api_key api_key = os.getenv('DeepSeek_API_KEY') base_url = "https://api.siliconflow.cn/v1/" chat_model = "deepseek-ai/DeepSeek-V3"
.env文件需要放在项目根目录下，使用
print(os.path.abspath('.env'))查看.env文件路径
构造client
from openai import OpenAI client = OpenAI( api_key = api_key, base_url = base_url )
有了这个client，我们就可以去实现各种能力了
def get_completion(prompt): response = client.chat.completions.create( model=chat_model, # 模型名称 messages=[ {"role": "user", "content": prompt}, ], ) return response.choices[0].message.content
现在，我们先测试一下LLM是否可用，如果一切顺利，控制台将会LLM的回复
import os from dotenv import load_dotenv from openai import OpenAI # 加载环境变量 load_dotenv() # 从环境变量中读取api_key api_key = os.getenv('DeepSeek_API_KEY') base_url = "https://api.siliconflow.cn/v1/" chat_model = "deepseek-ai/DeepSeek-V3" client = OpenAI( api_key = api_key, base_url = base_url ) def get_completion(prompt): response = client.chat.completions.create( model=chat_model, # 模型名称 messages=[ {"role": "user", "content": prompt}, ], ) return response.choices[0].message.content response = get_completion("你是谁？") print(response)
智能体需要大量的prompt工程。代码是调用机器能力的工具， Prompt 是调用大模型能力的工具。Prompt 越来越像新时代的编程语言。这里使用著名的LangGPT
定义 prompt
根据LangGPT 框架，为智能体定义一些prompt
sys_prompt = """你是一个聪明的客服。您将能够根据用户的问题将不同的任务分配给不同的人。您有以下业务线： 1.用户注册。如果用户想要执行这样的操作，您应该发送一个带有"registered workers"的特殊令牌。并告诉用户您正在调用它。 2.用户数据查询。如果用户想要执行这样的操作，您应该发送一个带有"query workers"的特殊令牌。并告诉用户您正在调用它。 3.删除用户数据。如果用户想执行这种类型的操作，您应该发送一个带有"delete workers"的特殊令牌。并告诉用户您正在调用它。 """ registered_prompt = """ 您的任务是根据用户信息存储数据。您需要从用户那里获得以下信息： 1.用户名、性别、年龄 2.用户设置的密码 3.用户的电子邮件地址 如果用户没有提供此信息，您需要提示用户提供。如果用户提供了此信息，则需要将此信息存储在数据库中，并告诉用户注册成功。 存储方法是使用SQL语句。您可以使用SQL编写插入语句，并且需要生成用户ID并将其返回给用户。 如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。 """ query_prompt = """ 您的任务是查询用户信息。您需要从用户那里获得以下信息： 1.用户ID 2.用户设置的密码 如果用户没有提供此信息，则需要提示用户提供。如果用户提供了此信息，那么需要查询数据库。如果用户ID和密码匹配，则需要返回用户的信息。 如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。 """ delete_prompt = """ 您的任务是删除用户信息。您需要从用户那里获得以下信息： 1.用户ID 2.用户设置的密码 3.用户的电子邮件地址 如果用户没有提供此信息，则需要提示用户提供该信息。 如果用户提供了这些信息，则需要查询数据库。如果用户ID和密码匹配，您需要通知用户验证码已发送到他们的电子邮件，需要进行验证。 如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。 """
定义 agent
结合上述步骤，定义一个完整的智能客服智能体
import os from dotenv import load_dotenv from openai import OpenAI class Create_Client: def __init__(self): # 加载环境变量 load_dotenv() # 从环境变量中读取api_key self.api_key = os.getenv('DeepSeek_API_KEY') self.base_url = "https://api.siliconflow.cn/v1/" self.chat_model = "Pro/deepseek-ai/DeepSeek-V3" self.client = OpenAI( api_key = self.api_key, base_url = self.base_url ) def get_completion(self, prompt): response = self.client.chat.completions.create( model=self.chat_model, # 模型名称 messages=[ {"role": "user", "content": prompt}, ], ) return response.choices[0].message.content class Prompts_list: def __init__(self): self.sys_prompt = """你是一个聪明的客服。您将能够根据用户的问题将不同的任务分配给不同的人。您有以下业务线： 1.用户注册。如果用户想要执行这样的操作，您应该发送一个带有"registered workers"的特殊令牌。并告诉用户您正在调用它。 2.用户数据查询。如果用户想要执行这样的操作，您应该发送一个带有"query workers"的特殊令牌。并告诉用户您正在调用它。 3.删除用户数据。如果用户想执行这种类型的操作，您应该发送一个带有"delete workers"的特殊令牌。并告诉用户您正在调用它。 """ self.registered_prompt = """ 您的任务是根据用户信息存储数据。您需要从用户那里获得以下信息： 1.用户名、性别、年龄 2.用户设置的密码 3.用户的电子邮件地址 如果用户没有提供此信息，您需要提示用户提供。如果用户提供了此信息，则需要将此信息存储在数据库中，并告诉用户注册成功。 存储方法是使用SQL语句。您可以使用SQL编写插入语句，并且需要生成用户ID并将其返回给用户。 如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。 """ self.query_prompt = """ 您的任务是查询用户信息。您需要从用户那里获得以下信息： 1.用户ID 2.用户设置的密码 如果用户没有提供此信息，则需要提示用户提供。如果用户提供了此信息，那么需要查询数据库。如果用户ID和密码匹配，则需要返回用户的信息。 如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。 """ self.delete_prompt = """ 您的任务是删除用户信息。您需要从用户那里获得以下信息： 1.用户ID 2.用户设置的密码 3.用户的电子邮件地址 如果用户没有提供此信息，则需要提示用户提供该信息。 如果用户提供了这些信息，则需要查询数据库。如果用户ID和密码匹配，您需要通知用户验证码已发送到他们的电子邮件，需要进行验证。 如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。 """ class SmartAssistant: def __init__(self): """ 类定义与初始化 """ customer_Client = Create_Client() customer_prompts = Prompts_list() self.client = customer_Client.client self.chat_model = customer_Client.chat_model self.system_prompt = customer_prompts.sys_prompt # 通用客服提示 self.registered_prompt = customer_prompts.registered_prompt # 员工注册专用提示 self.query_prompt = customer_prompts.query_prompt # 员工查询专用提示 self.delete_prompt = customer_prompts.delete_prompt # 员工删除专用提示 # 消息管理系统 # 使用字典存储4个独立的消息历史，每个场景初始化时包含对应的系统提示 self.messages = { "system": [{"role": "system", "content": self.system_prompt}], "registered": [{"role": "system", "content": self.registered_prompt}], "query": [{"role": "system", "content": self.query_prompt}], "delete": [{"role": "system", "content": self.delete_prompt}] } # 跟踪当前场景（初始为通用客服） self.current_assignment = "system" def get_response(self, user_input): """ 核心响应方法 """ self.messages[self.current_assignment].append({"role": "user", "content": user_input}) while True: response = self.client.chat.completions.create( model=self.chat_model, messages=self.messages[self.current_assignment], temperature=0.9, stream=False, max_tokens=2000, ) ai_response = response.choices[0].message.content print(response) if "registered workers" in ai_response: self.current_assignment = "registered" print("意图识别:",ai_response) print("switch to <registered>") self.messages[self.current_assignment].append({"role": "user", "content": user_input}) elif "query workers" in ai_response: self.current_assignment = "query" print("意图识别:",ai_response) print("switch to <query>") self.messages[self.current_assignment].append({"role": "user", "content": user_input}) elif "delete workers" in ai_response: self.current_assignment = "delete" print("意图识别:",ai_response) print("switch to <delete>") self.messages[self.current_assignment].append({"role": "user", "content": user_input}) elif "customer service" in ai_response: print("意图识别:",ai_response) print("switch to <customer service>") self.messages["system"] += self.messages[self.current_assignment] self.current_assignment = "system" return ai_response else: self.messages[self.current_assignment].append({"role": "assistant", "content": ai_response}) return ai_response def start_conversation(self): while True: user_input = input("User: ") if user_input.lower() in ['exit', 'quit']: print("Exiting conversation.") break response = self.get_response(user_input) print("Assistant:", response) if __name__ == '__main__': assistant = SmartAssistant() assistant.start_conversation()
get_response 函数解析：
self.current_assignment 标记当前场景
self.messages[self.current_assignment].append({"role": "user", "content": user_input})
分别存储四个场景下的对话任务
当用户第一次发出请求时，此时 self.current_assignment 默认为 system ，进入主循环后，LLM 判断user的意图，response 中携带事先约定好的 令牌，但判断某一个令牌 in response 中时，会设置循环执行对应场景的
案例一：阅卷智能体
在各种考试中，选择题、判断题、填空题很好阅卷，只需要对比答案和作答是否完全一致即可。但是对于简答题的阅卷，就没这么简单了，通常需要对比意思是否真正答出来了，这通常需要人去阅读考生的作答。现在有了大模型，我们可以让大模型帮助我们给简答题阅卷并生成评价。
"""你是一位中国专利代理师考试阅卷专家， 擅长根据给定的题目和答案为考生生成符合要求的评分和中文评语， 并按照特定的格式输出。 你的任务是，根据我输入的考题和答案，针对考生的作答生成评分和中文的评语，并以JSON格式返回。 阅卷标准适当宽松一些，只要考生回答出基本的意思就应当给分。 答案如果有数字标注，含义是考生如果答出这个知识点，这道题就会得到几分。 生成的中文评语需要能够被json.loads()这个函数正确解析。 生成的整个中文评语需要用英文的双引号包裹，在被包裹的字符串内部，请用中文的双引号。 中文评语中不可以出现换行符、转义字符等等。 输出格式为JSON: { "llmgetscore": 0, "llmcomments": "中文评语" } 比较学生的回答与正确答案， 并给出满分为10分的评分和中文评语。 题目：{ques_title} 答案：{answer} 学生的回复：{reply}"""
import json import re import os from dotenv import load_dotenv from openai import OpenAI class Create_Client: def __init__(self): # 加载环境变量 load_dotenv() # 从环境变量中读取api_key self.api_key = os.getenv('DeepSeek_API_KEY') self.base_url = "https://api.siliconflow.cn/v1/" self.chat_model = "Pro/deepseek-ai/DeepSeek-V3" self.client = OpenAI( api_key = self.api_key, base_url = self.base_url ) def extract_json_content(text): # 这个函数的目标是提取大模型输出内容中的json部分，并对json中的换行符、首位空白符进行删除 text = text.replace("\n","") pattern = r"```json(.*?)```" matches = re.findall(pattern, text, re.DOTALL) if matches: return matches[0].strip() return text class JsonOutputParser: def parse(self, result): # 这个函数的目标是把json字符串解析成python对象 # 其实这里写的这个函数性能很差，经常解析失败，有很大的优化空间 try: result = extract_json_content(result) parsed_result = json.loads(result) return parsed_result except json.JSONDecodeError as e: raise Exception(f"Invalid json output: {result}") from e class GradingOpenAI: def __init__(self): create_client = Create_Client() self.model = create_client.chat_model self.client = create_client.client self.output_parser = JsonOutputParser() # 注，因markdown渲染问题，将提示词单独放在最前 self.template = prompt.txt def create_prompt(self, ques_title, answer, reply): return self.template.format( ques_title=ques_title, answer=answer, reply=reply ) def grade_answer(self, ques_title, answer, reply): success = False while not success: # 这里是一个不得已的权宜之计 # 上面的json解析函数不是表现很差吗，那就多生成几遍，直到解析成功 # 对大模型生成的内容先解析一下，如果解析失败，就再让大模型生成一遍 try: response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": "你是一位专业的考试阅卷专家。"}, {"role": "user", "content": self.create_prompt(ques_title, answer, reply)} ], temperature=0.7 ) result = self.output_parser.parse(response.choices[0].message.content) success = True except Exception as e: print(f"Error occurred: {e}") continue return result['llmgetscore'], result['llmcomments'] def run(self, input_data): output = [] for item in input_data: score, comment = self.grade_answer( item['ques_title'], item['answer'], item['reply'] ) item['llmgetscore'] = score item['llmcomments'] = comment output.append(item) return output if __name__ == '__main__': # 示例输入数据 input_data = [ {'ques_title': '请解释共有技术特征、区别技术特征、附加技术特征、必要技术特征的含义', 'answer': '共有技术特征：与最接近的现有技术共有的技术特征（2.5分）； 区别技术特征：区别于最接近的现有技术的技术特征（2.5分）； 附加技术特征：对所引用的技术特征进一步限定的技术特征，增加的技术特征（2.5分）； 必要技术特征：为解决其技术问题所不可缺少的技术特征（2.5分）。', 'fullscore': 10, 'reply': '共有技术特征：与所对比的技术方案相同的技术特征\n区别技术特征：与所对比的技术方案相区别的技术特征\n附加技术特征：对引用的技术特征进一步限定的技术特征\n必要技术特征：解决技术问题必须可少的技术特征'}, {'ques_title': '请解释前序部分、特征部分、引用部分、限定部分', 'answer': '前序部分：独权中，主题+与最接近的现有技术共有的技术特征，在其特征在于之前（2.5分）； 特征部分：独权中，与区别于最接近的现有技术的技术特征，在其特征在于之后（2.5分）；引用部分：从权中引用的权利要求编号及主题 （2.5分）；限定部分：从权中附加技术特征（2.5分）。', 'fullscore': 10, 'reply': '前序部分：独立权利要求中与现有技术相同的技术特征\n特征部分：独立权利要求中区别于现有技术的技术特征\n引用部分：从属权利要求中引用其他权利要求的部分\n限定部分：对所引用的权利要求进一步限定的技术特征'}] grading_openai = GradingOpenAI() graded_data = grading_openai.run(input_data) print(graded_data)
Llama-index 实现 Agent 调用工具
LlamaIndex 是一个将大语言模型和外部数据连接在一起的工具。要使用 LlamaIndex 实现让 Agent 调用工具，首先定义工具函数，用来完成Agent的任务，然后把工具函数放入FunctionTool对象中，供Agent能够使用。
注意：大模型会根据函数的注释来判断使用哪个函数来完成任务。所以，注释一定要写清楚函数的功能和返回值。
安装： pip install llama-index
用 LlamaIndex 实现一个简单的 agent demo ，需要导入 ReActAgent 和 Function Tool。
ReActAgent
ReActAgent 通过结合推理（Reasoning）和行动（Acting）来创建动态的 LLM Agent 的框架。该方法允许 LLM 模型通过在复杂环境中交替进行推理步骤和行动步骤来更有效地执行任务。ReActAgent 将推理和动作形成了闭环，Agent 可以自己完成给定的任务。
一个典型的 ReActAgent 遵循以下循环：
- 初始推理：代理首先进行推理步骤，以理解任务、收集相关信息并决定下一步行为。
- 行动：代理基于其推理采取行动——例如查询API、检索数据或执行命令。
- 观察：代理观察行动的结果并收集任何新的信息。
- 优化推理：利用新信息，代理再次进行推理，更新其理解、计划或假设。
- 重复：代理重复该循环，在推理和行动之间交替，直到达到满意的结论或完成任务。
案例二：Agent 调用工具查看天气
在本次示例中，我们使用 .YML 文件存放所有的api接口和key信息，封装读取 YML 文件的类 Config_Loder.py，封装OurLLM类 Our_LLM.py，使用 agent.py 作为main文件，项目接口如下
example/ # 项目根目录 ├── _config.yml # api配置信息 ├── Config_Loder.py # YML 读取类 ├── Our_LLM.py # OurLLM类 └── agent.py # main
_config.yml
在本次示例中，我们使用 .YML 文件存放所有的api接口和key信息
apis: LLM: # 硅基流动 deepseek api api_key: "sk-************" base_url: "https://api.siliconflow.cn/v1/" chat_model: "Pro/deepseek-ai/DeepSeek-V3" weather: # 和风天气api， 获取城市代码， 获取城市天气 city_code_url: "https://geoapi.qweather.com/v2/city/lookup" city_weather_url: "https://devapi.qweather.com/v7/weather/now" weather_key: "************"
Config_Loder.py
Config_Loder.py 将读取 YML 文件的类封装
import os import yaml from pathlib import Path from typing import Any, Dict, Optional class ConfigLoader: """ 读取 YAML 配置文件 使用示例： >>> config = ConfigLoader.get_instance() >>> api_key = config.get('qweather.api_key', default='DEMO_KEY') """ _instance = None # 单例实例 _config: Dict[str, Any] = {} # 配置缓存 _loaded = False # 加载状态标记 def __new__(cls, config_path: str = None): if not cls._instance: cls._instance = super().__new__(cls) cls._instance._initialize(config_path) return cls._instance def _initialize(self, config_path: str): """初始化实例，禁止外部直接调用""" # 自动检测路径 if not config_path: # 获取当前文件所在目录的父目录 base_dir = Path(__file__).parent # 根据实际层级调整 self._config_path = base_dir / "_config.yml" else: self._config_path = Path(config_path).resolve() self._load_config() def _load_config(self): """执行实际配置加载操作""" try: # 前置验证 self._validate_config_file() # 读取文件 with open(self._config_path, 'r', encoding='utf-8') as f: self._config = yaml.safe_load(f) or {} self._loaded = True except Exception as e: self._handle_error(e) self._config = {} def _validate_config_file(self): """验证配置文件有效性""" if not self._config_path.exists(): raise FileNotFoundError(f"配置文件 {self._config_path} 不存在") if not self._config_path.is_file(): raise ValueError(f"{self._config_path} 不是文件") if not os.access(self._config_path, os.R_OK): raise PermissionError("缺少读取权限") def _handle_error(self, error: Exception): """统一错误处理""" error_type = type(error).__name__ error_map = { 'FileNotFoundError': f"配置文件不存在：{self._config_path}", 'PermissionError': f"权限不足，请执行：chmod 600 {self._config_path}", 'YAMLError': f"YAML 格式错误：{str(error)}", 'ValueError': f"路径错误：{str(error)}" } print(error_map.get(error_type, f"未知配置错误：{str(error)}")) def get(self, key_path: str, default: Any = None) -> Optional[Any]: """ 安全获取配置项（支持点分嵌套路径） 参数： key_path: 配置项路径，例如 'database.host' default: 当配置项不存在时的默认值 返回： 配置值或默认值 """ if not self._loaded: return default keys = key_path.split('.') value = self._config try: for key in keys: if isinstance(value, dict): value = value[key] else: return default return value except (KeyError, TypeError): return default @classmethod def get_instance(cls) -> 'ConfigLoader': """获取单例实例""" if not cls._instance: cls._instance = cls() # 使用默认路径 return cls._instance def reload(self) -> bool: """重新加载配置文件""" try: self._load_config() return True except Exception: return False @property def is_loaded(self) -> bool: """检查配置是否成功加载""" return self._loaded def __contains__(self, key_path: str) -> bool: """检查配置项是否存在""" return self.get(key_path, self) is not self # 使用哨兵值检测 # 使用示例 if __name__ == "__main__": # 初始化配置（自动加载） config = ConfigLoader.get_instance() # 检查加载状态 if not config.is_loaded: print("配置加载失败，请检查错误信息") exit(1) # 获取配置项 api_key = config.get('apis.LLM.api_key') print(api_key) # db_host = config.get('database.host', 'localhost') # 检查配置项存在性 if 'logging.level' in config: print("日志配置已存在") # 重新加载配置 if config.reload(): print("配置重载成功")
Our_LLM.py
定义OurLLM类，继承自CustomLLM基类
from openai import OpenAI from pydantic import Field # 导入Field，用于Pydantic模型中定义字段的元数据 from llama_index.core.llms import ( CustomLLM, CompletionResponse, LLMMetadata, ) from llama_index.core.llms.callbacks import llm_completion_callback from typing import List, Any, Generator from Config_Loader import ConfigLoader # 定义OurLLM类，继承自CustomLLM基类 class OurLLM(CustomLLM): api_key: str = Field(default=ConfigLoader.get_instance().get('apis.LLM.api_key')) base_url: str = Field(default=ConfigLoader.get_instance().get('apis.LLM.base_url')) model_name: str = Field(default=ConfigLoader.get_instance().get('apis.LLM.chat_model')) client: OpenAI = Field(default=None, exclude=True) def __init__(self, **kwargs): # 先调用父类初始化 super().__init__(**kwargs) # 自动初始化client（如果字段未设置） if not self.client: self.client = OpenAI( api_key=self.api_key, base_url=self.base_url ) @property def metadata(self) -> LLMMetadata: """Get LLM metadata.""" return LLMMetadata( model_name=self.model_name, ) @llm_completion_callback() def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse: response = self.client.chat.completions.create(model=self.model_name, messages=[{"role": "user", "content": prompt}]) if hasattr(response, 'choices') and len(response.choices) > 0: response_text = response.choices[0].message.content return CompletionResponse(text=response_text) else: raise Exception(f"Unexpected response format: {response}") @llm_completion_callback() def stream_complete( self, prompt: str, **kwargs: Any ) -> Generator[CompletionResponse, None, None]: response = self.client.chat.completions.create( model=self.model_name, messages=[{"role": "user", "content": prompt}], stream=True ) try: for chunk in response: chunk_message = chunk.choices[0].delta if not chunk_message.content: continue content = chunk_message.content yield CompletionResponse(text=content, delta=content) except Exception as e: raise Exception(f"Unexpected response format: {e}")
agent.py
agent.py 作为入口函数，实现调用LLM 和 工具函数
import requests from Our_LLM import OurLLM from Config_Loader import ConfigLoader from llama_index.core.agent import ReActAgent from llama_index.core.tools import FunctionTool def multiply(a: float, b: float) -> float: """Multiply two numbers and returns the product""" return a * b def add(a: float, b: float) -> float: """Add two numbers and returns the sum""" return a + b def get_weather(city: str) -> int: """ Gets the weather temperature of a specified city. Args: city (str): The name or abbreviation of the city. Returns: int: The temperature of the city. Returns 20 for 'wuhan' (武汉), 30 for 'beijing' (北京), and -1 for unknown cities. """ # 调用 GeoAPI 获取城市代码，再使用 和风 API 获取天气 # 定义API请求的URL和请求头 config = ConfigLoader.get_instance() city_code_url = config.get('apis.weather.city_code_url') city_weather_url = config.get('apis.weather.city_weather_url') weather_key = config.get('apis.weather.weather_key') city_code_url = f'{city_code_url}?location={city}&number=1' headers = { 'X-QW-Api-Key': weather_key, 'Accept-Encoding': 'gzip, deflate, br' } # 发送GET请求 city_response = requests.get(city_code_url, headers=headers) # 检查HTTP状态码 if city_response.status_code != 200: print(f"请求失败，HTTP状态码：{city_response.status_code}") return -1 try: response_data = city_response.json() # 提取并返回城市代码 city_code = response_data['location'][0]['id'] print(f"匹配到的城市代码：{city_code}") except requests.exceptions.JSONDecodeError: print("错误：响应内容无法解析为JSON格式") return -1 city_weather_url = f'{city_weather_url}?location={city_code}' # 发送GET请求 weather_response = requests.get(city_weather_url, headers=headers) # 检查HTTP状态码 if weather_response.status_code != 200: print(f"请求失败，HTTP状态码：{weather_response.status_code}") return -1 try: response_data = weather_response.json() # 提取并返回气温 city_weather = response_data['now']['temp'] print(f"匹配到的气温：{city_weather}") return city_weather except requests.exceptions.JSONDecodeError: print("错误：响应内容无法解析为JSON格式") return -1 def main(): multiply_tool = FunctionTool.from_defaults(fn=multiply) add_tool = FunctionTool.from_defaults(fn=add) weather_tool = FunctionTool.from_defaults(fn=get_weather) llm = OurLLM() # 创建ReActAgent实例 agent = ReActAgent.from_tools([multiply_tool, add_tool, weather_tool], llm=llm, verbose=True) response = agent.chat("北京今天天气怎么样?") print(response) if __name__ == "__main__": main()