1:2026年：为什么“会写Prompt“已经彻底不值钱？

# 2024年典型Prompt工程实现
import openai

class ContentGenerator:
    def __init__(self):
        self.client = openai.OpenAI()
        
    def generate_blog(self, topic, keywords):
        # 精心设计的Prompt模板
        prompt = f"""
        你是一位资深B2B技术内容营销专家。
        
        任务：撰写一篇关于{topic}的技术博客文章。
        
        要求：
        1. 目标受众：技术决策者（CTO、技术VP）
        2. 文章长度：1500-2000字
        3. 必须包含以下关键词：{', '.join(keywords)}
        4. 结构：引言→问题定义→解决方案→案例研究→结论
        5. 语气：专业、权威、有洞察力
        
        请直接输出文章内容，无需额外解释。
        """
        
        response = self.client.chat.completions.create(
            model="gpt-4-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=2500
        )
        
        return response.choices[0].message.content

# 2026年Agentic系统实现
from typing import List, Dict, Any
from dataclasses import dataclass
import openai
import json

@dataclass
class ContentTask:
    topic: str
    keywords: List[str]
    target_audience: str
    content_type: str  # blog, social, email
    brand_voice: str
    seo_requirements: Dict[str, Any]

class ContentMarketingAgent:
    """
    自进化内容营销Agent
    具备研究、规划、生成、优化、发布的完整闭环能力
    """
    
    def __init__(self):
        self.client = openai.OpenAI()
        self.memory = ContentMemory()  # 长期记忆系统
        self.performance_log = []  # 性能日志用于自反思
        
    def execute(self, task: ContentTask) -> Dict[str, Any]:
        """
        执行完整的内容生成工作流
        """
        # Step 1: 研究阶段 - 自主收集信息
        research_data = self._research_phase(task)
        
        # Step 2: 规划阶段 - 制定内容策略
        content_plan = self._planning_phase(task, research_data)
        
        # Step 3: 生成阶段 - 多模态内容创作
        draft_content = self._generation_phase(content_plan)
        
        # Step 4: 优化阶段 - SEO与品牌一致性检查
        optimized_content = self._optimization_phase(draft_content, task)
        
        # Step 5: 自反思 - 评估并记录性能
        self._reflection_phase(task, optimized_content)
        
        return {
            "content": optimized_content,
            "research_sources": research_data["sources"],
            "seo_score": optimized_content["seo_metrics"],
            "generation_metadata": self._get_metadata()
        }
    
    def _research_phase(self, task: ContentTask) -> Dict[str, Any]:
        """
        自主研究阶段：搜索最新资料、分析竞品内容、提取关键数据
        """
        # 调用搜索工具API
        search_results = self._call_search_tool(
            query=f"{task.topic} latest trends 2026",
            num_results=10
        )
        
        # 分析竞品内容
        competitor_analysis = self._analyze_competitors(task.topic)
        
        # 提取关键统计数据
        key_stats = self._extract_statistics(search_results)
        
        return {
            "sources": search_results,
            "competitor_insights": competitor_analysis,
            "statistics": key_stats
        }
    
    def _planning_phase(self, task: ContentTask, research: Dict) -> Dict[str, Any]:
        """
        规划阶段：制定详细的内容结构和策略
        """
        planning_prompt = f"""
        基于以下研究数据，制定内容策略：
        
        主题：{task.topic}
        目标受众：{task.target_audience}
        竞品分析：{json.dumps(research['competitor_insights'], ensure_ascii=False)}
        关键数据：{json.dumps(research['statistics'], ensure_ascii=False)}
        
        请输出JSON格式的内容规划：
        {{
            "headline_options": ["选项1", "选项2", "选项3"],
            "outline": ["章节1", "章节2", ...],
            "key_points": ["要点1", "要点2", ...],
            "cta_strategy": "转化策略",
            "estimated_word_count": 数字
        }}
        """
        
        response = self.client.chat.completions.create(
            model="gpt-5.4",
            messages=[{"role": "user", "content": planning_prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)
    
    def _generation_phase(self, plan: Dict) -> Dict[str, Any]:
        """
        生成阶段：基于规划生成高质量内容
        """
        # 分章节生成，保持连贯性
        sections = []
        context = ""
        
        for section_title in plan["outline"]:
            section_content = self._generate_section(
                title=section_title,
                context=context,
                key_points=plan["key_points"]
            )
            sections.append({
                "title": section_title,
                "content": section_content
            })
            context += f"\n\n{section_title}\n{section_content}"
        
        return {
            "sections": sections,
            "full_text": self._assemble_content(sections)
        }
    
    def _optimization_phase(self, content: Dict, task: ContentTask) -> Dict[str, Any]:
        """
        优化阶段：SEO检查、品牌声音一致性、可读性优化
        """
        # SEO评分
        seo_metrics = self._calculate_seo_score(
            content["full_text"],
            task.keywords,
            task.seo_requirements
        )
        
        # 品牌声音检查
        voice_consistency = self._check_brand_voice(
            content["full_text"],
            task.brand_voice
        )
        
        # 如果评分不达标，自动优化
        if seo_metrics["score"] < 85 or voice_consistency < 0.8:
            content["full_text"] = self._auto_optimize(
                content["full_text"],
                seo_metrics,
                task
            )
        
        return {
            "text": content["full_text"],
            "seo_metrics": seo_metrics,
            "voice_consistency": voice_consistency
        }
    
    def _reflection_phase(self, task: ContentTask, result: Dict):
        """
        自反思阶段：评估本次执行效果，更新策略
        """
        # 记录性能数据
        performance_entry = {
            "task": task,
            "result": result,
            "timestamp": datetime.now(),
            "success_metrics": self._evaluate_success(task, result)
        }
        
        self.performance_log.append(performance_entry)
        
        # 定期自训练：基于历史数据优化Prompt和策略
        if len(self.performance_log) % 10 == 0:
            self._self_improve()
    
    def _self_improve(self):
        """
        自进化机制：基于历史表现优化系统参数
        """
        # 分析高绩效任务的共同特征
        high_performers = [p for p in self.performance_log 
                          if p["success_metrics"]["conversion_rate"] > 0.05]
        
        # 更新策略参数
        if high_performers:
            self.memory.update_strategy(high_performers)

# 使用示例
agent = ContentMarketingAgent()
task = ContentTask(
    topic="Agentic AI在B2B营销中的应用",
    keywords=["Agentic AI", "B2B营销", "自动化", "2026趋势"],
    target_audience="B2B营销总监",
    content_type="blog",
    brand_voice="专业、前瞻、实用",
    seo_requirements={"min_word_count": 2000, "keyword_density": 0.02}
)

result = agent.execute(task)

from typing import Callable, Dict, Any
import json

class ToolCallingAgent:
    """
    工具调用Agent基础模板
    """
    
    def __init__(self):
        self.tools: Dict[str, Callable] = {}
        self.tool_schemas: Dict[str, Dict] = {}
    
    def register_tool(self, name: str, func: Callable, schema: Dict):
        """
        注册工具
        """
        self.tools[name] = func
        self.tool_schemas[name] = schema
    
    def execute(self, user_query: str) -> str:
        """
        执行用户查询，自动决定是否需要调用工具
        """
        # 构建系统Prompt
        system_prompt = self._build_system_prompt()
        
        # 第一次调用：决定是否需要工具
        response = self._call_llm(
            system=system_prompt,
            messages=[{"role": "user", "content": user_query}],
            tools=self._get_tool_definitions()
        )
        
        # 处理工具调用
        if response.tool_calls:
            tool_results = self._execute_tools(response.tool_calls)
            
            # 第二次调用：基于工具结果生成最终回答
            final_response = self._call_llm(
                system=system_prompt,
                messages=[
                    {"role": "user", "content": user_query},
                    {"role": "assistant", "content": response.content},
                    {"role": "tool", "content": json.dumps(tool_results)}
                ]
            )
            
            return final_response.content
        
        return response.content
    
    def _execute_tools(self, tool_calls: list) -> Dict[str, Any]:
        """
        执行工具调用
        """
        results = {}
        
        for call in tool_calls:
            tool_name = call.function.name
            tool_args = json.loads(call.function.arguments)
            
            if tool_name in self.tools:
                try:
                    result = self.tools[tool_name](**tool_args)
                    results[tool_name] = {"success": True, "data": result}
                except Exception as e:
                    results[tool_name] = {"success": False, "error": str(e)}
            else:
                results[tool_name] = {"success": False, "error": "Tool not found"}
        
        return results

"""
SelfEvolvingAgent - 自进化Agent基础框架
可用于内容生成、数据分析、代码辅助等场景
"""

from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json
import hashlib
from abc import ABC, abstractmethod

@dataclass
class Task:
    """任务定义"""
    id: str
    type: str
    input_data: Dict[str, Any]
    context: Dict[str, Any] = field(default_factory=dict)
    priority: int = 1

@dataclass
class ExecutionResult:
    """执行结果"""
    task_id: str
    success: bool
    output: Any
    metrics: Dict[str, float]
    timestamp: datetime
    reflection: Optional[str] = None

@dataclass
class Strategy:
    """策略定义"""
    name: str
    parameters: Dict[str, Any]
    success_rate: float = 0.0
    usage_count: int = 0

class Memory:
    """
    长期记忆系统
    存储历史执行记录和策略效果
    """
    
    def __init__(self):
        self.executions: List[ExecutionResult] = []
        self.strategies: Dict[str, Strategy] = {}
        self.patterns: Dict[str, Any] = {}
    
    def record_execution(self, result: ExecutionResult):
        """记录执行结果"""
        self.executions.append(result)
        self._update_strategy_stats(result)
    
    def get_similar_executions(self, task_type: str, limit: int = 10) -> List[ExecutionResult]:
        """获取相似任务的历史执行记录"""
        similar = [e for e in self.executions if e.task_id.startswith(task_type)]
        return sorted(similar, key=lambda x: x.timestamp, reverse=True)[:limit]
    
    def get_best_strategy(self, task_type: str) -> Optional[Strategy]:
        """获取某类任务的最佳策略"""
        relevant = [s for name, s in self.strategies.items() 
                   if name.startswith(task_type)]
        if relevant:
            return max(relevant, key=lambda x: x.success_rate)
        return None
    
    def _update_strategy_stats(self, result: ExecutionResult):
        """更新策略统计"""
        # 基于结果更新策略成功率
        pass

class Tool(ABC):
    """工具基类"""
    
    @abstractmethod
    def execute(self, **kwargs) -> Any:
        pass
    
    @abstractmethod
    def get_schema(self) -> Dict[str, Any]:
        pass

class SelfEvolvingAgent:
    """
    自进化Agent核心类
    
    特性：
    1. 工具调用能力
    2. 长期记忆系统
    3. 自反思机制
    4. 策略自优化
    """
    
    def __init__(self, llm_client, name: str = "Agent"):
        self.name = name
        self.llm = llm_client
        self.memory = Memory()
        self.tools: Dict[str, Tool] = {}
        self.strategies: Dict[str, Strategy] = {}
        self.reflection_threshold = 10  # 每10次执行触发一次深度反思
        self.execution_count = 0
    
    def register_tool(self, name: str, tool: Tool):
        """注册工具"""
        self.tools[name] = tool
    
    def execute(self, task: Task) -> ExecutionResult:
        """
        执行任务的主入口
        """
        self.execution_count += 1
        
        # 1. 检索记忆
        similar_cases = self.memory.get_similar_executions(task.type)
        best_strategy = self.memory.get_best_strategy(task.type)
        
        # 2. 规划执行
        plan = self._create_plan(task, similar_cases, best_strategy)
        
        # 3. 执行计划
        try:
            output = self._execute_plan(plan, task)
            success = True
            error_msg = None
        except Exception as e:
            output = None
            success = False
            error_msg = str(e)
        
        # 4. 计算指标
        metrics = self._calculate_metrics(task, output, success)
        
        # 5. 生成反思
        reflection = None
        if self.execution_count % self.reflection_threshold == 0:
            reflection = self._deep_reflection()
        elif not success:
            reflection = self._error_reflection(task, error_msg)
        
        # 6. 记录结果
        result = ExecutionResult(
            task_id=task.id,
            success=success,
            output=output,
            metrics=metrics,
            timestamp=datetime.now(),
            reflection=reflection
        )
        
        self.memory.record_execution(result)
        
        # 7. 触发自进化
        if self.execution_count % self.reflection_threshold == 0:
            self._self_evolve()
        
        return result
    
    def _create_plan(self, task: Task, similar_cases: List[ExecutionResult], 
                     best_strategy: Optional[Strategy]) -> Dict[str, Any]:
        """
        基于历史数据和最佳策略创建执行计划
        """
        context = {
            "task": task,
            "similar_cases": [self._case_to_dict(c) for c in similar_cases[:5]],
            "best_strategy": self._strategy_to_dict(best_strategy) if best_strategy else None,
            "available_tools": list(self.tools.keys())
        }
        
        plan_prompt = f"""
        你是一个自进化Agent的规划模块。
        
        任务信息：{json.dumps(context['task'].__dict__, default=str)}
        
        历史相似案例：{json.dumps(context['similar_cases'])}
        
        最佳策略：{json.dumps(context['best_strategy'])}
        
        可用工具：{json.dumps(context['available_tools'])}
        
        请输出JSON格式的执行计划：
        {{
            "steps": [
                {{
                    "order": 1,
                    "action": "工具名称或直接处理",
                    "input": {{}},
                    "expected_output": "描述"
                }}
            ],
            "fallback_strategy": "失败时的备选方案",
            "success_criteria": "判断成功的标准"
        }}
        """
        
        response = self.llm.chat.completions.create(
            model="gpt-5.4",
            messages=[{"role": "user", "content": plan_prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)
    
    def _execute_plan(self, plan: Dict, task: Task) -> Any:
        """
        执行计划
        """
        results = []
        
        for step in plan["steps"]:
            action = step["action"]
            
            if action in self.tools:
                # 调用工具
                tool_result = self.tools[action].execute(**step.get("input", {}))
                results.append({"step": step["order"], "tool": action, "result": tool_result})
            else:
                # 直接处理
                direct_result = self._direct_process(action, step.get("input", {}), task)
                results.append({"step": step["order"], "action": action, "result": direct_result})
        
        return results
    
    def _direct_process(self, action: str, input_data: Dict, task: Task) -> Any:
        """
        直接处理（不通过工具）
        """
        process_prompt = f"""
        执行以下处理步骤：
        
        动作：{action}
        输入：{json.dumps(input_data)}
        原始任务：{json.dumps(task.input_data)}
        
        请直接输出处理结果。
        """
        
        response = self.llm.chat.completions.create(
            model="gpt-5.4",
            messages=[{"role": "user", "content": process_prompt}]
        )
        
        return response.choices[0].message.content
    
    def _calculate_metrics(self, task: Task, output: Any, success: bool) -> Dict[str, float]:
        """
        计算执行指标
        """
        metrics = {
            "success": 1.0 if success else 0.0,
            "output_length": len(str(output)) if output else 0,
            "complexity_score": self._estimate_complexity(task)
        }
        
        return metrics
    
    def _estimate_complexity(self, task: Task) -> float:
        """
        估计任务复杂度
        """
        # 基于输入数据量和类型估计复杂度
        input_size = len(json.dumps(task.input_data))
        return min(input_size / 1000, 10.0)
    
    def _error_reflection(self, task: Task, error_msg: str) -> str:
        """
        错误反思
        """
        reflection_prompt = f"""
        执行失败，请进行反思：
        
        任务：{json.dumps(task.input_data)}
        错误：{error_msg}
        
        请分析失败原因并提出改进建议。
        """
        
        response = self.llm.chat.completions.create(
            model="gpt-5.4",
            messages=[{"role": "user", "content": reflection_prompt}]
        )
        
        return response.choices[0].message.content
    
    def _deep_reflection(self) -> str:
        """
        深度反思：基于最近N次执行进行系统性反思
        """
        recent_executions = self.memory.executions[-self.reflection_threshold:]
        
        success_count = sum(1 for e in recent_executions if e.success)
        success_rate = success_count / len(recent_executions)
        
        reflection_prompt = f"""
        进行深度反思：
        
        最近{self.reflection_threshold}次执行统计：
        - 成功率：{success_rate:.2%}
        - 平均指标：{self._avg_metrics(recent_executions)}
        
        失败案例分析：
        {self._format_failures(recent_executions)}
        
        请分析系统表现，识别改进机会，并提出策略优化建议。
        """
        
        response = self.llm.chat.completions.create(
            model="gpt-5.4",
            messages=[{"role": "user", "content": reflection_prompt}]
        )
        
        return response.choices[0].message.content
    
    def _self_evolve(self):
        """
        自进化：基于反思结果优化策略
        """
        # 分析历史数据，识别高效策略
        strategy_analysis = self._analyze_strategies()
        
        # 更新策略库
        for strategy_name, stats in strategy_analysis.items():
            if strategy_name in self.strategies:
                self.strategies[strategy_name].success_rate = stats["success_rate"]
                self.strategies[strategy_name].usage_count = stats["usage_count"]
        
        # 生成新策略变体
        new_strategies = self._generate_strategy_variants()
        for strategy in new_strategies:
            self.strategies[strategy.name] = strategy
    
    def _analyze_strategies(self) -> Dict[str, Dict]:
        """
        分析策略效果
        """
        # 实现策略分析逻辑
        return {}
    
    def _generate_strategy_variants(self) -> List[Strategy]:
        """
        生成策略变体
        """
        # 实现策略生成逻辑
        return []
    
    def _case_to_dict(self, case: ExecutionResult) -> Dict:
        """转换案例为字典"""
        return {
            "task_id": case.task_id,
            "success": case.success,
            "metrics": case.metrics,
            "reflection": case.reflection
        }
    
    def _strategy_to_dict(self, strategy: Strategy) -> Dict:
        """转换策略为字典"""
        return {
            "name": strategy.name,
            "parameters": strategy.parameters,
            "success_rate": strategy.success_rate
        }
    
    def _avg_metrics(self, executions: List[ExecutionResult]) -> Dict[str, float]:
        """计算平均指标"""
        if not executions:
            return {}
        
        all_metrics = [e.metrics for e in executions]
        avg = {}
        
        for key in all_metrics[0].keys():
            values = [m.get(key, 0) for m in all_metrics]
            avg[key] = sum(values) / len(values)
        
        return avg
    
    def _format_failures(self, executions: List[ExecutionResult]) -> str:
        """格式化失败案例"""
        failures = [e for e in executions if not e.success]
        return json.dumps([{"task": e.task_id, "reflection": e.reflection} for e in failures])


# ==================== 使用示例 ====================

class SearchTool(Tool):
    """搜索工具示例"""
    
    def execute(self, query: str, num_results: int = 5) -> Dict:
        # 实际实现中调用搜索API
        return {
            "query": query,
            "results": [f"Result {i} for {query}" for i in range(num_results)]
        }
    
    def get_schema(self) -> Dict:
        return {
            "name": "search",
            "description": "搜索网络信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "num_results": {"type": "integer", "default": 5}
                },
                "required": ["query"]
            }
        }

# 初始化Agent
# agent = SelfEvolvingAgent(llm_client=openai.OpenAI())
# agent.register_tool("search", SearchTool())

# 创建任务
# task = Task(
#     id="research_001",
#     type="research",
#     input_data={"topic": "Agentic AI trends 2026"}
# )

# 执行任务
# result = agent.execute(task)