泰安市网站建设_网站建设公司_前后端分离_seo优化

实战背景：智能体「行动后反思」的自动化：我们如何让系统从错误日志中生成改进用例

概述

本项目是一个基于智能体「行动后反思」的自动化：我们如何让系统从错误日志中生成改进用例观点和方法实现的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, tools=None): """封装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. 模拟模式设计

程序支持两种运行模式：

1. 实际模式：使用真实的DeepSeek API
2. 模拟模式：在没有有效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_key=API_KEY, base_url=BASE_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 f"The weather in {city} is Sunny, 25°C." def calculate_tax(amount): """计算税费工具""" # 模拟：工具只接受整数，如果 Agent 传入字符串数字可能导致逻辑错误 try: val = int(amount) return f"Tax 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, tools=None): """封装 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( id="call_mock_12345", function=SimpleNamespace( name="calculate_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( id="call_mock_67890", function=SimpleNamespace( name="get_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_calls=None ) return mock_message # 实际调用LLM response = client.chat.completions.create( model="deepseek-chat", messages=messages, tools=tools, temperature=0.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_ascii=False)} 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( model="deepseek-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_success=False 或 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 Agent's system prompt or tool definitions. Reflection Unit: {json.dumps(reflection.__dict__, ensure_ascii=False)} 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( model="deepseek-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__, indent=2, ensure_ascii=False)) # 3. 闭环：生成改进建议 # 如果 Agent 没能正确处理任务，这里会生成改进建议 improvement = reviewer.generate_improvement_case(reflection) if improvement: print(f"\n💡 [Improvement Case] (可直接用于工程迭代):") print(json.dumps(improvement.__dict__, indent=2, ensure_ascii=False)) 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.11

2. 安装依赖包：

pip install openai

配置API密钥

通过环境变量

export DEEPSEEK_API_KEY="你的DeepSeek API密钥"

运行程序

确保已安装依赖后，使用以下命令运行程序：

模拟模式运行：

python main.py

程序会自动使用模拟模式运行，无需配置API密钥。

查看结果

程序运行后会显示：

1. Agent执行过程的详细日志

2. 生成的Reflection Unit

3. 生成的Improvement Case（如果有）

程序运行用户手册

程序流程

1.任务接收：Agent接收用户任务"Please calculate the tax for $500 dollars."

2.任务规划：调用LLM判断是否需要使用工具

3.工具调用：决定调用calculate_tax工具处理税务计算

4.结果处理：获取工具执行结果并生成最终回答

5.反思生成：基于执行过程生成Reflection Unit

6.改进建议：根据反思结果生成Improvement Case

输出说明

• `[Agent] 开始执行任务`：Agent开始处理任务

• `[Agent] 决定调用工具`：Agent决定调用的工具列表

• `[System] 正在生成反思单元`：系统正在生成反思信息

• `[System] 任务成功`：任务执行成功，无需改进建议

• `[Reflection Unit]`：结构化的反思信息

• `[Improvement Case]`：工程改进建议

未来优化点

1. 功能增强

- 支持更多类型的工具和服务

- 实现更复杂的任务规划和分解能力

- 添加多轮对话支持

2. 反思系统优化

- 增强Reflection Unit的分析维度

- 实现Improvement Case的自动应用机制

- 添加回归测试集生成功能

结果与展示

正常流程

错误流程(反思+自动化回归测试用例)