网站文章图片如何跳转,游戏软件开发专业,站长工具查询网站,河南省住房和城乡建设厅投诉网站实战背景#xff1a;智能体「行动后反思」的自动化#xff1a;我们如何让系统从错误日志中生成改进用例
概述
本项目是一个基于智能体「行动后反思」的自动化#xff1a;我们如何让系统从错误日志中生成改进用例观点和方法实现的Python智能体Demo#xff0c;旨在验证智能…实战背景智能体「行动后反思」的自动化我们如何让系统从错误日志中生成改进用例概述本项目是一个基于智能体「行动后反思」的自动化我们如何让系统从错误日志中生成改进用例观点和方法实现的Python智能体Demo旨在验证智能体的核心概念Agent的规划、工具路由和调用、参数解析等都通过LLM能力实现实现了行动后反思系统(Post-Action Review System)将执行日志转化为结构化资产(Reflection Unit和Improvement Case)核心概念1.智能体(Agent)核心能力LLM驱动决策Agent的所有决策都由LLM完成包括任务规划、工具选择和参数解析工具调用通过LLM判断何时需要调用工具以及调用哪个工具自主执行Agent能够自主执行工具调用并处理结果2. 行动后反思系统Reflection Unit(反思单元)将执行过程结构化为标准格式便于分析Improvement Case(改进用例)基于反思单元自动生成的工程改进建议自动化质量保证系统自动识别潜在问题并提出改进方案程序架构sample2/ ├── main.py # 主程序文件 ├── TEACHING_DOCUMENT.md # 教学文档 └── requirements.txt # 依赖文件程序主要包含以下组件1.SimpleAgent类实现智能体的核心功能2.PostMortemSystem类实现行动后反思系统3.Mock工具模拟外部服务调用4.模拟模式在没有有效API密钥时使用模拟响应重点代码解读1. SimpleAgent类class SimpleAgent: def init (self): self.history [] # 对话历史记录 self.logs [] # 执行日志 def call_llm(self, messages, toolsNone): 封装LLM调用支持模拟模式 # 在模拟模式下返回预设响应 # 在实际模式下调用DeepSeek API def execute(self, user_query): 执行任务的主循环 # 1. 初始化对话历史 # 2. 调用LLM进行任务规划和工具选择 # 3. 执行选定的工具 # 4. 根据工具结果生成最终回答2. PostMortemSystem类class PostMortemSystem: def generate_reflection_unit(self, user_query, logs, final_result): 生成反思单元 # 分析执行日志生成结构化的反思信息 def generate_improvement_case(self, reflection): 生成改进用例 # 基于反思单元生成工程改进建议3. 模拟模式设计程序支持两种运行模式实际模式使用真实的DeepSeek API模拟模式在没有有效API密钥时自动切换便于教学和测试主程序main.py# -*- coding: utf-8 -*- import os import json import time from typing import List, Dict, Any, Optional try: from openai import OpenAI HAS_OPENAI True except ImportError: HAS_OPENAI False # # 0. 配置与基础结构 # # 从环境变量获取 DeepSeek API Key如果没有设置则使用默认值 API_KEY os.getenv(DEEPSEEK_API_KEY, sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx) BASE_URL https://api.deepseek.com # 只在有openai包且API密钥有效时才初始化client if HAS_OPENAI and API_KEY ! sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx: client OpenAI(api_keyAPI_KEY, base_urlBASE_URL) USE_MOCK False else: client None USE_MOCK True print([INFO] 使用模拟模式运行因为未配置有效的API密钥) class ReflectionUnit: 文章中定义的结构化反思单元 def __init__(self, task_goal, action_plan, tools_used, expected_outcome, actual_outcome, error_type, confidence, is_success): self.task_goal task_goal self.action_plan action_plan self.tools_used tools_used self.expected_outcome expected_outcome self.actual_outcome actual_outcome self.error_type error_type self.confidence confidence self.is_success is_success class ImprovementCase: 文章中定义的改进用例 def __init__(self, failure_pattern, trigger_condition, suggested_change, risk_level): self.failure_pattern failure_pattern self.trigger_condition trigger_condition self.suggested_change suggested_change self.risk_level risk_level # # 1. Mock Tools (模拟工具层) # def get_weather(city): 查询天气工具 # 模拟故意让某个城市报错制造“可学习的错误” if Unknown in city: return Error: City not found in database. return fThe weather in {city} is Sunny, 25°C. def calculate_tax(amount): 计算税费工具 # 模拟工具只接受整数如果 Agent 传入字符串数字可能导致逻辑错误 try: val int(amount) return fTax for {val} is {val * 0.1} except ValueError: return Error: Input must be a valid integer. TOOLS_DEF [ { type: function, function: { name: get_weather, description: Get weather for a city, parameters: { type: object, properties: {city: {type: string}}, required: [city] } } }, { type: function, function: { name: calculate_tax, description: Calculate 10% tax for an amount, parameters: { type: object, properties: {amount: {type: integer}}, required: [amount] } } } ] # # 2. Agent Core (智能体核心) # class SimpleAgent: def __init__(self): self.history [] self.logs [] # 原始日志 def call_llm(self, messages, toolsNone): 封装 LLM 调用 # 如果使用模拟模式则使用模拟响应 if USE_MOCK or not HAS_OPENAI or client is None: print([Mock Mode] 模拟LLM调用使用预设响应...) # 根据消息内容决定模拟响应 user_content for msg in messages: if msg.get(role) user: user_content msg.get(content, ) break # 模拟工具调用决策 if tools and (tax in user_content.lower() or calculate in user_content.lower()): # 模拟决定调用calculate_tax工具 from types import SimpleNamespace mock_tool_call SimpleNamespace( idcall_mock_12345, functionSimpleNamespace( namecalculate_tax, arguments{amount: 500} ) ) mock_message SimpleNamespace( content, tool_calls[mock_tool_call] ) return mock_message elif tools and (weather in user_content.lower()): # 模拟决定调用get_weather工具 from types import SimpleNamespace mock_tool_call SimpleNamespace( idcall_mock_67890, functionSimpleNamespace( nameget_weather, arguments{city: Beijing} ) ) mock_message SimpleNamespace( content, tool_calls[mock_tool_call] ) return mock_message else: # 模拟普通回复 from types import SimpleNamespace mock_message SimpleNamespace( content这是一个模拟的LLM回复。, tool_callsNone ) return mock_message # 实际调用LLM response client.chat.completions.create( modeldeepseek-chat, messagesmessages, toolstools, temperature0.0 ) return response.choices[0].message def execute(self, user_query): 执行任务的主循环 print(f\n [Agent] 开始执行任务: {user_query}) self.history [{role: system, content: You are a helpful assistant. Use tools when necessary.}, {role: user, content: user_query}] # 记录原始日志 self.logs.append({step: init, query: user_query}) # Step 1: 规划与工具调用 msg self.call_llm(self.history, TOOLS_DEF) self.history.append(msg) tool_calls msg.tool_calls final_result if tool_calls: print(f️ [Agent] 决定调用工具: {[t.function.name for t in tool_calls]}) self.logs.append({step: tool_decision, tools: [t.function.name for t in tool_calls]}) for tool in tool_calls: func_name tool.function.name args json.loads(tool.function.arguments) # 路由执行 result Error if func_name get_weather: result get_weather(**args) elif func_name calculate_tax: result calculate_tax(**args) print(f - 工具输出: {result}) self.history.append({ role: tool, tool_call_id: tool.id, content: str(result) }) self.logs.append({step: tool_execution, tool: func_name, args: args, result: result}) # Step 2: 根据工具结果生成最终回答 final_msg self.call_llm(self.history) final_result final_msg.content else: final_result msg.content print(f [Agent] 最终结果: {final_result}) self.logs.append({step: final_response, content: final_result}) return final_result # # 3. Post-Action Review System (行动后反思系统) # class PostMortemSystem: 实现文章中的核心观点将日志转化为结构化资产 def generate_reflection_unit(self, user_query, logs, final_result): 核心方法生成反思单元 不依赖人工而是让 LLM 基于 Logs 进行结构化总结 print(\n [System] 正在生成反思单元 (Reflection Unit)...) # 如果使用模拟模式则使用模拟响应 if USE_MOCK or not HAS_OPENAI or client is None: print([Mock Mode] 模拟生成反思单元...) # 创建模拟的反思单元 mock_data { task_goal: user_query, action_plan: 调用calculate_tax工具处理税务计算, tools_used: [calculate_tax], expected_outcome: 正确计算$500的税费, actual_outcome: 工具调用成功返回税费计算结果, error_type: None, confidence: 0.95, is_success: True } return ReflectionUnit(**mock_data) # 构造 Meta-Prompt让 LLM 扮演“复盘专家” prompt f You are a QA Engineer for an AI Agent system. Analyze the following execution logs and generate a structured Reflection Unit. User Goal: {user_query} Final Output: {final_result} Execution Logs: {json.dumps(logs, ensure_asciiFalse)} Return a JSON object strictly following this structure: {{ task_goal: Summarized goal, action_plan: What the agent planned to do, tools_used: [list of tools], expected_outcome: What should have happened ideally, actual_outcome: What actually happened, error_type: One of [None, Decision Error, Parameter Error, Understanding Bias], confidence: 0.0 to 1.0 (self-assessment of success), is_success: true/false }} response client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], response_format{type: json_object} ) data json.loads(response.choices[0].message.content) print(data) return ReflectionUnit(**data) def generate_improvement_case(self, reflection): 核心方法生成改进用例 只有当检测到错误is_successFalse 或 error_type ! None时才触发 if reflection.is_success and reflection.error_type None: print(✅ [System] 任务成功无需生成改进用例。) return None # 如果使用模拟模式则使用模拟响应 if USE_MOCK or not HAS_OPENAI or client is None: print([Mock Mode] 模拟生成改进用例...) # 创建模拟的改进用例仅在有错误时 if not reflection.is_success or reflection.error_type ! None: mock_data { failure_pattern: 参数解析错误, trigger_condition: 当用户输入带符号的金额时, suggested_change: 改进Agent的参数提取逻辑自动去除货币符号, risk_level: low } return ImprovementCase(**mock_data) else: return None print(\n⚠️ [System] 检测到失败/偏差正在生成改进用例 (Improvement Case)...) prompt f Based on the following Reflection Unit, generate an engineering Improvement Case. We need to fix the Agents system prompt or tool definitions. Reflection Unit: {json.dumps(reflection.__dict__, ensure_asciiFalse)} Return JSON: {{ failure_pattern: Abstract pattern of this failure, trigger_condition: When does this happen?, suggested_change: Specific engineering change (Prompt/Router/Tool), risk_level: low/medium/high }} response client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], response_format{type: json_object} ) data json.loads(response.choices[0].message.content) return ImprovementCase(**data) # # 4. 主程序入口 # def run_agent_task(agent, reviewer, task): 执行单个Agent任务 print(f\n{*50}) print(f 接收到任务: {task}) print(f{*50}) # 1. Agent 执行 result agent.execute(task) # 2. 自动化反思 (Post-Action Review) # 无论成功与否系统都会生成 Reflection Unit reflection reviewer.generate_reflection_unit(task, agent.logs, result) print(f\n [Reflection Unit]:) print(json.dumps(reflection.__dict__, indent2, ensure_asciiFalse)) # 3. 闭环生成改进建议 # 如果 Agent 没能正确处理任务这里会生成改进建议 improvement reviewer.generate_improvement_case(reflection) if improvement: print(f\n [Improvement Case] (可直接用于工程迭代):) print(json.dumps(improvement.__dict__, indent2, ensure_asciiFalse)) print(\n--- 这里的 suggested_change 应该被自动推送到 Prompt 优化队列或回归测试集中。) def main(): 主程序入口 print( 欢迎使用智能体演示程序) print( * 50) # 实例化 agent SimpleAgent() reviewer PostMortemSystem() # 检查是否有命令行参数 import sys if len(sys.argv) 1: # 如果有命令行参数直接执行该任务 task .join(sys.argv[1:]) run_agent_task(agent, reviewer, task) return # 交互模式 while True: print(\n 请输入您的问题输入 quit 或 exit 退出程序:) print( 示例问题:) print( • 请问北京的天气怎么样) print( • 请计算500元的税费) print( • 请计算$50i0美元的税费测试错误处理) try: task input(\n 请输入问题: ).strip() # 检查退出条件 if task.lower() in [quit, exit, 退出]: print(\n 感谢使用智能体演示程序再见) break # 检查空输入 if not task: print(⚠️ 输入不能为空请重新输入。) continue # 执行任务 run_agent_task(agent, reviewer, task) # 询问是否继续 print(\n *50) continue_choice input( 是否继续测试其他问题(y/n): ).strip().lower() if continue_choice in [n, no, 否, quit, exit]: print(\n 感谢使用智能体演示程序再见) break except KeyboardInterrupt: print(\n\n 程序被用户中断再见) break except Exception as e: print(f\n❌ 程序运行出错: {str(e)}) print( 请重新输入问题或联系开发者。) if __name__ __main__: main()使用说明环境准备1. 确保已安装Python 3.112. 安装依赖包pip install openai配置API密钥通过环境变量export DEEPSEEK_API_KEY你的DeepSeek API密钥运行程序确保已安装依赖后使用以下命令运行程序模拟模式运行python main.py程序会自动使用模拟模式运行无需配置API密钥。查看结果程序运行后会显示1. Agent执行过程的详细日志2. 生成的Reflection Unit3. 生成的Improvement Case如果有程序运行用户手册程序流程1.任务接收Agent接收用户任务Please calculate the tax for $500 dollars.2.任务规划调用LLM判断是否需要使用工具3.工具调用决定调用calculate_tax工具处理税务计算4.结果处理获取工具执行结果并生成最终回答5.反思生成基于执行过程生成Reflection Unit6.改进建议根据反思结果生成Improvement Case输出说明[Agent] 开始执行任务Agent开始处理任务[Agent] 决定调用工具Agent决定调用的工具列表[System] 正在生成反思单元系统正在生成反思信息[System] 任务成功任务执行成功无需改进建议[Reflection Unit]结构化的反思信息[Improvement Case]工程改进建议未来优化点1. 功能增强- 支持更多类型的工具和服务- 实现更复杂的任务规划和分解能力- 添加多轮对话支持2. 反思系统优化- 增强Reflection Unit的分析维度- 实现Improvement Case的自动应用机制- 添加回归测试集生成功能结果与展示正常流程错误流程(反思自动化回归测试用例)