1. 项目概述:当Python遇见Gemini AI,构建你的智能体
最近在捣鼓AI应用开发,发现一个挺有意思的趋势:大家不再满足于简单的问答机器人,而是开始构建能自主思考、执行复杂任务的“智能体”。正好,Google的Gemini AI模型开放了API,功能强大,尤其是其多模态和长上下文能力,简直是构建智能体的绝佳“大脑”。于是,我花了几周时间,用Python深度整合Gemini AI,搭建了一个功能相对完整的AI-Agent原型。这个项目不是简单的API调用,而是涉及了任务规划、工具调用、记忆管理和自主决策循环等核心概念。如果你对如何让AI从“聊天”升级到“做事”感兴趣,或者想用Python和最新的Gemini模型搞点有意思的自动化应用,那这篇从零到一的实战记录或许能给你一些启发。
简单来说,这个项目就是用Python作为“身体”和“神经系统”,Gemini AI作为“决策大脑”,创建一个能够理解复杂指令、拆解任务、调用工具(比如搜索网络、读写文件、执行代码)并最终完成目标的智能程序。它适合有一定Python基础,对AI应用开发、自动化流程或者智能助手构建感兴趣的开发者。即使你对Agent的概念还比较模糊,跟着步骤走一遍,也能清晰理解其内部工作机制。
2. 核心架构与设计思路拆解
2.1 为什么选择Python + Gemini AI这个组合?
构建AI-Agent,选型是第一步。我选择Python,原因很直接:生态无敌。无论是处理各种数据格式(JSON, CSV)、进行网络请求(requests库)、操作本地系统(os, subprocess),还是集成成千上万的专业库(如pandas数据分析,selenium网页操控),Python都有成熟、简洁的解决方案。Agent的本质是“协调者”,它需要灵活地调用各种工具,Python在这方面是当之无愧的“胶水语言”。
而选择Gemini AI作为核心模型,则经过了多方面的考量。首先,多模态能力是刚需。一个真正的智能体不应该只懂文字。Gemini原生支持文本、图像、音频、视频的理解和生成。这意味着你的Agent可以分析用户上传的图表截图并总结数据,或者处理一段语音指令。其次,超长的上下文窗口(Gemini 1.5 Pro支持百万级token)至关重要。Agent在执行任务过程中会产生大量的历史对话、工具调用结果和内部思考过程,这些都需要作为上下文喂给模型,以保持任务的连贯性和一致性。短上下文模型很容易“遗忘”之前的步骤。最后,Google在函数调用(Function Calling)上的支持越来越完善。虽然OpenAI的GPT系列在此领域先行一步,但Gemini的API也提供了稳定的工具调用接口,能够清晰地将模型生成的参数结构化地传递给Python函数,这是实现Agent自主操作的关键。
2.2 智能体的核心组件与工作流设计
一个基础的、可运行的AI-Agent,我认为至少需要四个核心组件,它们共同构成了一个闭环的工作流。
1. 规划与决策模块(大脑):这是Gemini模型的核心职责。它接收用户的指令和当前上下文(包括记忆和历史),分析任务意图,并将其分解成一系列可执行的子步骤。例如,用户说“帮我分析一下公司上个月的销售数据,并总结成一份PPT报告”。大脑需要规划出:“步骤1:定位或请求销售数据文件;步骤2:读取并分析数据,找出关键指标;步骤3:根据分析结果,生成PPT的文字大纲和图表描述;步骤4:调用PPT生成工具,将大纲和描述转化为实际文件。”
2. 工具集(双手):这是一系列Python函数,每个函数代表Agent能做的具体事情。比如:search_web(query)用于搜索最新信息,read_file(file_path)用于读取文档,execute_python_code(code_string)用于运行计算或数据处理脚本,generate_image(prompt)用于文生图等。这些工具的定义(函数名、描述、参数列表及类型)需要以特定的格式(通常是JSON Schema)提供给Gemini模型,这样模型才知道“手”能做什么、怎么做。
3. 记忆系统(海马体):Agent需要有短期记忆和长期记忆。短期记忆即当前会话的上下文,通常通过维护一个对话历史列表来实现。长期记忆则更复杂,可能涉及向量数据库(如ChromaDB, Pinecone),用于存储和检索过往的重要交互信息,让Agent在多次对话中保持“个性”和知识连续性。在本项目中,我们先实现一个基于对话轮次的简单记忆管理。
4. 执行与协调引擎(脑干与脊髓):这是一个Python主循环程序。它负责串联起以上所有组件:初始化模型、加载工具、维护记忆;将用户输入和记忆提交给“大脑”获取决策;解析决策结果,识别出需要调用的工具;安全地执行对应的工具函数;将工具执行结果反馈给“大脑”进行下一轮思考;循环直至任务完成或达到终止条件。
这个工作流可以概括为:“思考(Plan)-> 行动(Act)-> 观察(Observe)”的循环。Agent不断思考下一步该做什么(调用哪个工具、传入什么参数),执行行动,观察结果(工具返回的数据),然后将结果纳入上下文进行下一轮思考,直到最终答案产生。
3. 环境搭建与核心依赖详解
3.1 基础环境与Gemini API配置
首先,确保你的Python环境是3.9以上版本。我强烈建议使用虚拟环境来管理依赖,避免包冲突。
# 创建并激活虚拟环境(以venv为例) python -m venv agent_env source agent_env/bin/activate # Linux/macOS # 或 agent_env\Scripts\activate # Windows接下来安装核心库。除了官方的Google Generative AI SDK,我们还需要一些辅助库。
pip install google-generativeai # 核心,用于调用Gemini模型 pip install python-dotenv # 管理环境变量,安全存储API密钥 pip install requests # 用于实现网络搜索等工具 pip install chromadb # 可选,用于实现向量记忆存储获取并配置Gemini API密钥:这是最关键的一步。前往Google AI Studio,创建一个项目并生成API密钥。千万不要将密钥硬编码在代码中!
在你的项目根目录创建一个名为.env的文件,内容如下:
GOOGLE_API_KEY=你的_实际_API_密钥_放在这里然后在Python代码中这样加载:
import os import google.generativeai as genai from dotenv import load_dotenv load_dotenv() # 从.env文件加载环境变量 GOOGLE_API_KEY = os.getenv('GOOGLE_API_KEY') if not GOOGLE_API_KEY: raise ValueError("请在 .env 文件中设置 GOOGLE_API_KEY") genai.configure(api_key=GOOGLE_API_KEY)注意:API调用是收费的,虽然新用户有免费额度,但在开发调试阶段,尤其是使用流式响应或复杂任务时,建议在代码中设置合理的速率限制,并关注AI Studio控制台的使用量。
3.2 初始化Gemini模型与关键参数选择
Gemini提供了多个模型,对于Agent开发,gemini-1.5-pro是目前功能最全面、上下文最长的选择,适合作为“大脑”。gemini-1.5-flash速度更快,成本更低,适合对响应速度要求高、任务相对简单的场景。
# 选择模型 model_name = 'gemini-1.5-pro' # 或 'gemini-1.5-flash' model = genai.GenerativeModel(model_name) # 配置生成参数,这对Agent的“思考”质量影响很大 generation_config = { "temperature": 0.7, # 创造性:0.0更确定/保守,1.0更随机/有创意。Agent任务建议0.7-0.9以鼓励探索。 "top_p": 0.95, # 核采样:与temperature配合,控制词汇选择的集中程度。 "top_k": 40, # 采样范围:只从概率最高的k个词中选。0表示禁用。 "max_output_tokens": 2048, # 单次响应最大长度。对于复杂规划,可以设得大一些。 "response_mime_type": "application/json", # 强制返回JSON格式,便于解析工具调用 }参数详解与选择逻辑:
- temperature:这是最重要的参数之一。对于需要严谨规划、工具调用的Agent任务,不宜过高(如1.2),否则模型可能产生天马行空、参数不准确的工具调用。也不宜过低(如0.1),否则会过于死板,缺乏应对复杂、模糊指令的能力。0.7-0.9是一个较好的平衡区间,既能保证一定的推理能力,又能保持输出的稳定性。
- response_mime_type:设置为
”application/json”是一个高级技巧。我们可以要求模型严格按照我们定义的JSON格式来返回它的“思考”和“决策”,这极大简化了后续的解析流程。你需要为模型提供相应的JSON结构示例。
4. 核心模块实现:打造智能体的“躯干”与“工具手”
4.1 定义与封装工具函数
工具是Agent能力的延伸。每个工具都应该是一个纯Python函数,并有清晰的元数据描述。我们先实现几个最常用的工具。
import json import subprocess import sys import requests from datetime import datetime # 工具1:执行Python代码(沙盒环境,实际应用需更安全) def execute_python_code(code: str) -> str: """ 执行一段Python代码并返回结果或错误信息。 注意:这是一个简化版,生产环境需要严格的沙盒隔离。 """ try: # 使用subprocess在独立进程中运行,限制时间和输出 result = subprocess.run( [sys.executable, "-c", code], capture_output=True, text=True, timeout=30 ) if result.returncode == 0: return f"代码执行成功。输出:\n{result.stdout}" else: return f"代码执行出错。错误信息:\n{result.stderr}" except subprocess.TimeoutExpired: return "错误:代码执行超时(30秒)。" except Exception as e: return f"执行过程发生未知错误:{str(e)}" # 工具2:网络搜索(使用DuckDuckGo即时答案或模拟请求) def search_web(query: str) -> str: """ 根据查询词进行网络搜索,返回简洁的摘要信息。 这里使用一个公开的API示例,实际可使用Serper、SerpAPI等专业服务。 """ # 示例:使用DuckDuckGo的HTML抓取(简易版,可能不稳定) try: url = f"https://api.duckduckgo.com/" params = { "q": query, "format": "json", "no_html": "1", "skip_disambig": "1" } response = requests.get(url, params=params, timeout=10) data = response.json() abstract = data.get('AbstractText', '') if abstract: return f"搜索 '{query}' 的结果摘要:{abstract}" else: # 如果没有摘要,返回相关主题 related = data.get('RelatedTopics', []) topics = [topic.get('Text', '')[:100] for topic in related[:3] if topic.get('Text')] return f"未找到直接摘要。相关主题:{'; '.join(topics)}" if topics else "未找到相关信息。" except Exception as e: return f"网络搜索失败:{str(e)}" # 工具3:获取当前时间 def get_current_time(*args) -> str: # args参数是为了统一接口,即使不用也要保留 """ 返回当前的日期和时间。 """ now = datetime.now() return now.strftime("当前时间是:%Y年%m月%d日 %H时%M分%S秒") # 工具字典:将函数名与函数对象、元数据关联起来 TOOLS = { "execute_python_code": { "function": execute_python_code, "description": "执行一段Python代码字符串。用于计算、数据处理或自动化脚本。", "parameters": { "code": {"type": "string", "description": "要执行的Python代码字符串"} } }, "search_web": { "function": search_web, "description": "在互联网上搜索信息。适用于获取最新新闻、事实核查或未知领域知识。", "parameters": { "query": {"type": "string", "description": "搜索查询关键词"} } }, "get_current_time": { "function": get_current_time, "description": "获取当前的系统日期和时间。", "parameters": {} # 此工具无需参数 } }工具设计心得:
- 单一职责:每个工具只做一件事,并且做好。函数名和描述要清晰准确,这直接影响了模型能否正确理解和使用它。
- 安全第一:像
execute_python_code这样的工具是“危险”的。上述实现仅用于演示,绝对不可用于生产环境或暴露给不受信任的用户。生产环境需要使用Docker容器、资源限制、系统调用白名单等严格隔离措施。 - 健壮性:工具函数内部要有完善的异常处理(try-except),返回的字符串信息要友好且包含关键错误详情,以便模型能理解哪里出错了,并调整策略。
- 元数据标准化:我们用了简单的字典来存储工具元数据。更规范的做法是使用Pydantic模型来定义,确保类型安全。
4.2 构建模型提示词与工具描述系统
模型需要知道它有哪些工具可用。我们需要将工具描述转换成模型能理解的格式,并设计一个结构化的提示词来引导模型的行为。
def build_tools_description() -> str: """将工具字典转换为给模型看的自然语言描述和JSON Schema格式。""" descriptions = [] for tool_name, tool_info in TOOLS.items(): desc = f"- {tool_name}: {tool_info['description']}" params = tool_info.get('parameters', {}) if params: param_desc = [] for param_name, param_info in params.items(): param_desc.append(f" - {param_name} ({param_info['type']}): {param_info['description']}") desc += "\n 参数:\n" + "\n".join(param_desc) descriptions.append(desc) return "\n".join(descriptions) def build_system_prompt() -> str: """构建系统提示词,定义Agent的角色、能力和输出格式。""" tools_desc = build_tools_description() prompt = f""" 你是一个智能AI助手,能够通过调用工具来帮助用户完成复杂任务。 你拥有以下工具: {tools_desc} **你的工作流程**: 1. 理解用户请求。 2. 思考需要如何一步步解决,可能需要调用工具。 3. 如果需要调用工具,你必须严格按照以下JSON格式回应: ```json {{ "thought": "你的思考过程,解释为什么选择这个工具以及参数如何确定", "action": {{ "name": "工具名称", "args": {{"参数名1": "参数值1", "参数名2": "参数值2"}} }} }}- 如果你认为已经收集到足够信息,可以直接回答用户问题,则用以下格式:
{{ "thought": "你的最终思考总结", "final_answer": "给用户的最终回答" }}重要规则:
- 一次只调用一个工具。
- 工具参数必须与描述中定义的类型相符。
- 如果工具执行失败,根据错误信息调整你的计划。
- 你的所有回应必须是且仅是上述两种JSON格式之一,不要添加任何其他解释性文字。 """ return prompt
这个系统提示词是Agent的“宪法”,它规定了模型的思考框架和输出格式。使用JSON格式强制输出,使得我们的Python程序可以稳定地解析出`action`或`final_answer`,这是实现自动化循环的关键。 ### 4.3 实现主执行循环与状态管理 现在,我们将所有部分串联起来,实现核心的“思考-行动”循环。 ```python import json class SimpleAgent: def __init__(self, model): self.model = model self.conversation_history = [] # 短期记忆:存储所有消息 self.system_prompt = build_system_prompt() # 初始化对话,注入系统提示 self.conversation_history.append({"role": "user", "parts": [self.system_prompt]}) # 这里通常需要一个模型的“假响应”来设定上下文,但Gemini的ChatSession处理方式不同。 # 我们采用另一种方式:在每次生成时,都将系统提示和历史拼接起来。 def _format_history_for_model(self) -> list: """将对话历史格式化为Gemini API所需的格式。""" # Gemini的`generate_content`可以接受一个parts列表,我们将历史和系统提示合并。 # 更优的方案是使用`genai.ChatSession`来管理多轮对话。 formatted_messages = [] # 加入系统提示作为第一条用户消息 formatted_messages.append({"role": "user", "parts": [self.system_prompt]}) for msg in self.conversation_history[1:]: # 跳过第一条(已经是系统提示) # 注意:这里简化了角色映射。实际Gemini API中,parts可以是文本或函数调用结果。 formatted_messages.append(msg) return formatted_messages def process_user_input(self, user_input: str) -> str: """处理用户输入,运行Agent循环,返回最终答案或中间状态。""" # 1. 将用户输入加入历史 self.conversation_history.append({"role": "user", "parts": [user_input]}) max_turns = 10 # 防止无限循环 for turn in range(max_turns): # 2. 准备当前上下文:系统提示 + 完整历史 # 由于我们每次都需要系统提示,这里简单地将系统提示和历史文本拼接。 # 更高效的做法是使用ChatSession,这里为清晰起见采用拼接。 context_parts = [self.system_prompt] for msg in self.conversation_history[1:]: # 从第一条用户输入开始 role_prefix = "用户" if msg["role"] == "user" else "助手" context_parts.append(f"{role_prefix}: {msg['parts'][0]}") context_text = "\n\n".join(context_parts) + "\n\n助手:" # 3. 调用模型进行“思考” try: response = self.model.generate_content( context_text, generation_config=generation_config ) response_text = response.text.strip() except Exception as e: return f"调用模型时出错:{str(e)}" # 4. 尝试解析模型的JSON响应 try: # 清理响应文本,可能包含markdown代码块标记 if response_text.startswith("```json"): response_text = response_text[7:] # 去掉 ```json if response_text.endswith("```"): response_text = response_text[:-3] decision = json.loads(response_text) except json.JSONDecodeError: # 如果模型没有返回合法JSON,将其视为最终回答(可能是直接对话) final_answer = response_text self.conversation_history.append({"role": "model", "parts": [final_answer]}) return final_answer # 5. 判断决策类型:调用工具 or 最终回答 if "final_answer" in decision: # 模型决定给出最终答案 final_answer = decision["final_answer"] self.conversation_history.append({"role": "model", "parts": [final_answer]}) return final_answer elif "action" in decision: # 模型决定调用工具 action = decision["action"] tool_name = action.get("name") args = action.get("args", {}) # 6. 执行工具 if tool_name not in TOOLS: tool_result = f"错误:未知工具 '{tool_name}'。可用工具有:{list(TOOLS.keys())}" else: tool_func = TOOLS[tool_name]["function"] try: # 调用工具函数 tool_result = tool_func(**args) except TypeError as e: tool_result = f"错误:调用工具参数不匹配。{str(e)}" except Exception as e: tool_result = f"错误:工具执行过程中发生异常。{str(e)}" # 7. 将工具执行结果(观察)加入历史,供下一轮思考使用 result_message = f"工具调用结果:{tool_result}" self.conversation_history.append({"role": "model", "parts": [response_text]}) # 模型的思考/决策 self.conversation_history.append({"role": "user", "parts": [result_message]}) # 将结果作为“用户”输入反馈 print(f"[Turn {turn+1}] 思考:{decision.get('thought')}") print(f"[Turn {turn+1}] 行动:调用 {tool_name},参数 {args}") print(f"[Turn {turn+1}] 结果:{tool_result[:100]}...") # 打印部分结果 else: # 响应格式不符合预期 error_msg = f"错误:无法解析模型的响应格式。响应内容:{response_text}" self.conversation_history.append({"role": "model", "parts": [error_msg]}) return error_msg # 循环超过最大轮次 return f"任务未在{max_turns}轮内完成。最后状态:{self.conversation_history[-1]}"这个SimpleAgent类封装了核心逻辑。初始化时载入系统提示和历史。process_user_input方法是引擎,它:
- 接收用户输入。
- 将历史和当前输入组合成完整上下文。
- 调用Gemini模型,要求其输出JSON格式的决策。
- 解析决策:如果是最终答案,则返回;如果是工具调用,则找到对应函数执行。
- 将工具执行结果作为新一轮的“用户输入”反馈给模型,形成循环。
5. 实战演示:让Agent处理复杂任务
让我们用一个具体的例子,看看这个Agent是如何工作的。假设用户提出一个需要多步骤处理的任务。
# 初始化Agent agent = SimpleAgent(model) # 任务1:一个需要计算和获取信息的复合任务 user_query_1 = “现在几点了?另外,帮我计算一下15的阶乘是多少。” print(“用户提问:”, user_query_1) result_1 = agent.process_user_input(user_query_1) print(“\n最终回答:”, result_1) print(“-” * 50) # 重置对话历史,开始新任务(在实际中,可能不清空,以实现连续对话) agent.conversation_history = [{"role": "user", "parts": [agent.system_prompt]}] # 任务2:一个需要搜索和推理的任务 user_query_2 = “特斯拉(Tesla)最新的电动车型是什么?它的续航里程大概是多少公里?” print(“用户提问:”, user_query_2) result_2 = agent.process_user_input(user_query_2) print(“\n最终回答:”, result_2)预期执行流程分析(以任务1为例):
- 第一轮思考:模型收到指令“现在几点了?另外,帮我计算一下15的阶乘是多少。”。它分析出这是两个独立问题,但可以顺序解决。它决定先调用
get_current_time工具,因为无需参数,且能立即回答第一个问题。- 模型输出JSON:
{"thought": “用户问了两个问题。我先回答第一个关于时间的问题,这需要调用get_current_time工具。然后我再处理第二个计算问题。”, "action": {"name": "get_current_time", "args": {}}}
- 模型输出JSON:
- 第一轮行动:Agent解析JSON,调用
get_current_time()函数,获得类似“当前时间是:2024年05月27日 14时30分15秒”的结果。 - 第一轮观察:将结果反馈给模型上下文。
- 第二轮思考:模型收到时间结果,并记得还有第二个问题。它决定调用
execute_python_code工具来计算15的阶乘。- 模型输出JSON:
{"thought": “第一个问题已回答。现在需要计算15的阶乘。我将使用Python的math.factorial函数来完成这个计算。”, "action": {"name": "execute_python_code", "args": {"code": "import math\\nresult = math.factorial(15)\\nprint(result)"}}}
- 模型输出JSON:
- 第二轮行动:Agent执行代码,得到结果
1307674368000。 - 第二轮观察:将代码执行成功的结果反馈给模型。
- 第三轮思考:模型认为已获得回答两个问题所需的全部信息,可以组织最终答案。
- 模型输出JSON:
{"thought": “我已经获得了当前时间和15的阶乘计算结果。现在可以将这两个信息整合成一个完整、友好的回答给用户。”, "final_answer": “当前时间是2024年05月27日 14时30分15秒。\\n15的阶乘是 1,307,674,368,000。”}
- 模型输出JSON:
- 结束:Agent收到
final_answer,将其返回给用户,循环结束。
通过控制台打印的[Turn x]信息,你可以清晰地看到这个“思考-行动-观察”的循环过程。对于任务2,模型可能会先调用search_web工具获取最新信息,然后再组织答案。
6. 性能优化与高级功能拓展
基础循环跑通后,我们可以从多个维度提升这个Agent的实用性、效率和智能水平。
6.1 优化提示工程与思维链引导
原始的提示词可能不够强大。我们可以引入更先进的提示技术:
- Few-Shot示例:在系统提示中,直接给模型提供几个“用户提问-模型正确思考与调用过程”的完整示例。这能极大地提升模型输出格式的准确性和任务分解的合理性。
- 思维链(Chain-of-Thought)强化:要求模型在
”thought”字段中进行更详细、逐步的推理。例如:“用户的目标是Y。要达成Y,需要先完成A和B。完成A需要工具X,其参数可能是…因为…。完成B需要…”。详细的思考过程不仅使调试更方便,也常常能提高最终决策的质量。 - 输出格式严格化:使用JSON Schema在提示词中精确描述所需的输出格式,甚至可以使用Gemini API的
response_mime_type=”application/json”配合response_schema参数进行强制约束,这比在自然语言中描述更可靠。
6.2 实现长期记忆与向量检索
简单的对话历史是短期记忆。要实现“记住过去对话”的能力,需要引入向量数据库。
- 存储:每次对话结束后,将重要的用户查询和Agent的最终回答(或关键信息片段)转换为文本嵌入(Embedding)。可以使用Gemini的嵌入模型
embedding-001或text-embedding-004。 - 检索:当新对话开始时,将当前用户查询也转换为嵌入,然后在向量数据库中搜索最相关的历史片段(基于余弦相似度)。
- 注入上下文:将检索到的相关历史信息,作为“长期记忆”插入到本次对话的系统提示或上下文开头。
# 伪代码示例 import chromadb from chromadb.config import Settings # 初始化向量数据库客户端 chroma_client = chromadb.Client(Settings(persist_directory="./chroma_db")) collection = chroma_client.get_or_create_collection(name="agent_memory") def store_memory(text: str, metadata: dict): """将一段文本存入长期记忆""" embedding = genai.embed_content(model="models/embedding-001", content=text)['embedding'] collection.add( embeddings=[embedding], documents=[text], metadatas=[metadata], ids=[f"id_{int(time.time())}"] ) def retrieve_memory(query: str, n_results=3) -> list: """根据查询检索相关记忆""" query_embedding = genai.embed_content(model="models/embedding-001", content=query)['embedding'] results = collection.query( query_embeddings=[query_embedding], n_results=n_results ) return results['documents'][0] # 返回相关的文本列表这样,当用户问“我们上次讨论的那个项目进展如何?”时,Agent就能检索到相关的历史对话,实现上下文关联。
6.3 并行工具调用与流程优化
目前的Agent是严格串行的:思考 -> 调用一个工具 -> 观察 -> 再思考。对于一些彼此独立的任务,可以优化为并行调用。
实现思路:修改提示词,允许模型在单次思考中规划多个可并行执行的动作,输出一个动作列表。然后,Agent主循环使用concurrent.futures线程池并发执行这些工具调用,等待所有结果返回后,一次性汇总反馈给模型进行下一步思考。这能显著减少需要多步独立数据获取任务的总耗时。
6.4 安全性、稳定性与成本控制
- 工具调用沙盒化:如前所述,对
execute_python_code、shell_command等危险工具必须进行沙盒隔离。可以使用docker run --rm -v在临时容器中运行代码,并严格限制CPU、内存、运行时间和网络访问。 - 循环超时与中断:设置最大轮次(如20轮)和总时间限制(如120秒)。当Agent陷入死循环或无效尝试时,能够自动终止并返回当前最佳结果或错误信息。
- Token消耗监控:Gemini API按输入输出Token数计费。在
conversation_history中积累过多内容会导致上下文过长,成本激增。需要实现策略:要么定期总结历史并压缩(让模型自己总结之前的对话),要么只保留最近N轮对话,将更早的对话存入向量记忆,需要时再检索。 - 错误处理与降级:网络请求、API调用都可能失败。代码中应有重试机制(如
tenacity库)。如果某个工具连续失败,应能将其从本次任务中“禁用”,并通知模型调整方案。
7. 常见问题与调试技巧实录
在开发过程中,我遇到了不少坑,这里总结一下,希望能帮你绕过去。
7.1 模型不按JSON格式输出
这是初期最常见的问题。模型可能会输出纯文本,或者在JSON外添加额外解释。
- 解决方案1:强化提示词。在系统提示的开头和结尾反复强调“你必须只输出JSON”,“你的回应必须是有效的JSON,不要有任何其他文字”。使用```json代码块包裹示例。
- 解决方案2:使用API的强制JSON模式。Gemini API支持
response_mime_type=”application/json”,并可以传入response_schema参数(一个JSON Schema对象)。这是最可靠的方法,能从根本上保证输出格式。 - 解决方案3:后处理与容错。在解析响应时,使用更健壮的提取方法,比如用正则表达式匹配
{...}之间的内容,或者使用json5库(比标准json更宽松)来解析。同时,做好降级处理,如果解析失败,尝试将整个响应作为最终答案或错误信息。
7.2 工具参数类型或值错误
模型可能理解了要调用哪个工具,但给出的参数值类型不对(比如要求数字却给了字符串),或者值本身不合理。
- 解决方案1:在提示词中明确参数类型。在工具描述里,不仅写“参数名”,一定要写明“类型”,如
(string),(integer),(boolean)。 - 解决方案2:在Python代码中进行验证和转换。在调用工具函数前,对参数进行类型检查和转换。例如,如果参数应该是整数,尝试
int(args[‘param’]),如果失败则返回清晰错误给模型。 - 解决方案3:提供更详细的参数描述。不要只写“查询关键词”,可以写成“查询关键词,应尽可能具体,例如‘2024年巴黎奥运会开幕式时间’而非‘奥运会信息’”。
7.3 Agent陷入无效循环或原地打转
有时Agent会反复调用同一个工具,或者在不同的工具间来回切换,无法推进任务。
- 解决方案1:在思考中引入“禁止”指令。在反馈给模型的工具结果中,如果发现重复调用,可以加上“注意:你刚刚已经调用过这个工具并得到了结果,请基于已有信息进行下一步分析,不要重复相同的操作。”
- 解决方案2:丰富工具的结果信息。工具返回的结果要尽可能详尽和有信息量。例如,网络搜索工具在没找到结果时,不要只返回“未找到”,可以返回“未找到关于‘XXX’的明确信息。建议尝试更换关键词为‘YYY’或‘ZZZ’进行搜索。”,给模型提供调整方向的线索。
- 解决方案3:设置轮次限制和超时。这是最后的保障。并在接近限制时,提醒模型“你已经进行了很多步,请尝试总结当前已有信息,直接给出你能得出的最佳答案。”
7.4 上下文长度管理与成本优化
长上下文是双刃剑,既带来强大连贯性,也带来高昂成本。
- 技巧1:选择性记忆。不是所有对话历史都需要原封不动地放入上下文。可以只保留最近3-5轮完整对话,对于更早的,在每一轮Agent思考完成后,让其自己用一句话总结“当前任务状态和已获得的关键信息”,然后用这个总结替换掉之前冗长的原始交互记录。
- 技巧2:分层上下文。将上下文分为“系统指令层”、“短期记忆层(最近几轮)”、“长期记忆摘要层”和“关键信息层(如当前任务目标)”。模型主要关注短期记忆和关键信息,长期记忆摘要仅在需要时(通过向量检索)注入。
- 技巧3:使用更经济的模型处理上下文。对于只需要理解上下文而不需要复杂规划的任务,可以使用
gemini-1.5-flash来处理历史总结或信息提取,然后将结果交给gemini-1.5-pro进行核心决策。
构建一个真正鲁棒、实用的AI-Agent是一个持续迭代的过程。从最简单的循环开始,逐步增加记忆、优化提示、完善工具、加强安全。这个用Python和Gemini AI搭建的框架,已经为你提供了一个坚实的起点。接下来,你可以根据特定领域(比如数据分析Agent、客服Agent、个人知识管理Agent)去丰富工具库,打磨提示词,让它真正成为你得力的数字助手。