智能体开发实战：从零构建一个能够自主决策的Agent系统（保姆级讲解）

class Agent:
    def __init__(self):
        self.memory = []  # 记忆模块
        self.tools = {}   # 工具模块
        self.plan = []    # 规划模块
        
    def think(self, task):
        # 思考模块：分析任务，制定计划
        pass
    
    def act(self, step):
        # 行动模块：执行具体步骤
        pass
    
    def observe(self, result):
        # 观察模块：收集反馈
        pass

import openai
import json
from typing import List, Dict, Any

class ResearchAgent:
    def __init__(self, api_key: str, model: str = "gpt-4"):
        self.client = openai.OpenAI(api_key=api_key)
        self.model = model
        self.memory = []
        self.max_steps = 10
        
    def think(self, query: str) -> Dict[str, Any]:
        """决策下一步行动"""
        prompt = f"""
        任务：研究"{query}"
        
        历史步骤：
        {self._format_memory()}
        
        可选行动：
        1. search_web: 搜索网络信息
        2. analyze_data: 分析已收集数据
        3. generate_report: 生成研究报告
        
        请以JSON格式返回决策结果：
        {{
            "action": "选择的行动",
            "reason": "选择原因",
            "parameters": {{"参数名": "参数值"}}
        }}
        """
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        
        return json.loads(response.choices[0].message.content)

class ToolRegistry:
    @staticmethod
    def search_web(query: str, num_results: int = 5) -> List[str]:
        """模拟网络搜索功能"""
        # 实际项目中可集成SerpAPI、Google Search API等
        search_results = [
            f"关于{query}的最新研究进展...",
            f"{query}的核心技术原理...",
            f"{query}在实际应用中的案例分析..."
        ]
        return search_results[:num_results]
    
    @staticmethod
    def analyze_data(data: List[str], focus: str) -> str:
        """分析收集到的数据"""
        # 调用LLM进行数据分析
        analysis_prompt = f"""
        请分析以下数据，重点关注{focus}：
        {data}
        
        请提炼关键观点、趋势和矛盾点。
        """
        # 这里简化处理，实际应调用LLM
        return f"分析完成，发现{focus}方面的3个关键趋势..."

class MemoryModule:
    def __init__(self, capacity: int = 100):
        self.short_term = []  # 短期记忆
        self.long_term = {}   # 长期记忆
        self.capacity = capacity
        
    def add_experience(self, thought: str, action: str, result: str):
        """添加经验到记忆"""
        experience = {
            "thought": thought,
            "action": action,
            "result": result,
            "timestamp": time.time()
        }
        
        self.short_term.append(experience)
        
        # 经验总结，存入长期记忆
        if len(self.short_term) >= self.capacity:
            self._consolidate_memory()
    
    def _consolidate_memory(self):
        """将短期记忆整合为长期记忆"""
        # 使用LLM对短期记忆进行总结
        summary = "根据多次执行经验，总结出的最佳实践..."
        self.long_term[time.time()] = summary
        self.short_term = []

def advanced_think(self, query: str) -> Dict[str, Any]:
    """带思维链和自我反思的决策"""
    
    # 第一步：问题分解
    decomposition_prompt = f"""
    请将任务"{query}"分解为3-5个子任务：
    考虑：哪些子任务是必须的？执行顺序应该是怎样的？
    """
    
    # 第二步：评估可用资源
    resource_prompt = "当前可用的工具有：搜索、数据分析、报告生成"
    
    # 第三步：风险评估
    risk_prompt = "每个步骤可能遇到什么困难？如何规避？"
    
    # 第四步：最终决策
    # 将以上思考整合，做出最优决策
    
    # 第五步：反思与调整
    if self.memory:
        reflection = self._reflect_on_previous_actions()
        if reflection["needs_adjustment"]:
            return self._adjust_plan(reflection["suggestions"])

def execute_with_adaptation(self, plan: List[Dict[str, Any]]):
    """自适应执行计划"""
    for step in plan:
        max_retries = 3
        for attempt in range(max_retries):
            try:
                result = self._execute_step(step)
                self._record_success(step, result)
                break
            except Exception as e:
                if attempt == max_retries - 1:
                    # 彻底失败，修改计划
                    step = self._modify_step(step, str(e))
                else:
                    # 重试前等待
                    time.sleep(2 ** attempt)
        
        # 检查是否需要调整后续步骤
        if self._should_adjust_plan(result):
            plan = self._adjust_remaining_plan(plan, result)

agent = ResearchAgent(api_key="your-api-key")
result = agent.run("大模型在金融领域的应用现状与未来趋势")

print(result)