【LLM应用开发原理】我怎样理解Agent

class AgentHierarchy:
    """Agent的层次结构"""
    
    # Level 1: 交互层
    class InteractionLayer:
        """直接与用户交互的界面"""
        def handle_input(self, user_request):
            # 意图识别
            # 输入验证
            # 前置处理
            pass
        
        def render_output(self, response):
            # 格式化输出
            # 流式渲染
            # 错误处理
            pass
    
    # Level 2: 推理层
    class ReasoningLayer:
        """基于LLM的核心推理能力"""
        def plan(self, goal, context):
            # 任务分解
            # 决策制定
            # 思维链生成
            pass
        
        def reflect(self, action, result):
            # 反思与调整
            # 学习反馈
            pass
    
    # Level 3: 执行层
    class ExecutionLayer:
        """与环境交互的能力"""
        def execute(self, action):
            # 工具调用
            # 状态管理
            # 错误恢复
            pass
        
        def observe(self):
            # 环境感知
            # 状态更新
            pass
    
    # Level 4: 记忆层
    class MemoryLayer:
        """持久化知识存储"""
        def store(self, key, value):
            # 短期记忆
            # 长期记忆
            pass
        
        def retrieve(self, query):
            # 语义检索
            # 关联推荐
            pass

# 没有Agent的LLM：被困在文本世界
llm = LargeLanguageModel()
answer = llm.ask("帮我写一个数据分析报告")
# 结果：只能生成文本建议，无法实际操作

# 有Agent的LLM：连接真实世界
agent = DataAnalysisAgent()
result = agent.execute("帮我分析Q1销售数据并生成导出报告")
# 结果：连接数据库 → 执行分析 → 生成图表 → 发送邮件

class IntentRouter:
    """意图路由器"""
    
    def __init__(self):
        self.classifier = IntentClassifier()  # 可以是规则引擎或小型LLM
        self.handlers = {
            'search': SearchHandler(),
            'analysis': AnalysisHandler(),
            'generation': GenerationHandler(),
            'action': ActionHandler(),
            'conversation': ConversationHandler()
        }
    
    def route(self, user_input: str, context: dict) -> Handler:
        """路由到合适的处理器"""
        # 步骤1：提取意图
        intent = self.classifier.classify(user_input, context)
        
        # 步骤2：检查是否有对应的处理器
        if intent not in self.handlers:
            intent = 'conversation'  # 降级到通用对话
        
        # 步骤3：返回处理器
        return self.handlers[intent]


class IntentClassifier:
    """意图分类器"""
    
    def classify(self, text: str, context: dict) -> str:
        """使用混合策略分类"""
        # 策略1：关键词匹配（快速路径）
        if any(word in text.lower() for word in ['搜索', '查找', 'search']):
            return 'search'
        
        if any(word in text.lower() for word in ['分析', '统计', '分析']):
            return 'analysis'
        
        # 策略2：语义相似度（中等路径）
        intent = self._semantic_match(text)
        if intent:
            return intent
        
        # 策略3：LLM分类（慢速但准确路径）
        return self._llm_classify(text, context)

class Planner:
    """任务规划器"""
    
    def plan(self, goal: str, context: dict, available_tools: list) -> Plan:
        """生成执行计划"""
        
        # 步骤1：理解目标
        goal_understanding = self._understand_goal(goal, context)
        
        # 步骤2：分解子任务
        subtasks = self._decompose(goal_understanding, available_tools)
        
        # 步骤3：确定执行顺序
        execution_order = self._order_subtasks(subtasks)
        
        # 步骤4：生成完整计划
        return Plan(
            goal=goal,
            steps=execution_order,
            estimated_cost=self._estimate_cost(execution_order),
            risk_level=self._assess_risk(execution_order)
        )
    
    def _understand_goal(self, goal: str, context: dict) -> dict:
        """深入理解用户目标"""
        prompt = f"""
        分析以下用户目标，提取关键信息：
        
        目标: {goal}
        上下文: {context}
        
        请分析并返回JSON格式：
        {{
            "intent": "核心意图",
            "scope": "任务范围",
            "constraints": ["约束条件1", "约束条件2"],
            "preferences": ["偏好1", "偏好2"],
            "success_criteria": ["成功标准1", "成功标准2"]
        }}
        """
        return self.llm.ask(prompt, response_format="json")
    
    def _decompose(self, goal_understanding: dict, tools: list) -> list:
        """将目标分解为子任务"""
        prompt = f"""
        将以下目标分解为可执行的子任务：
        
        目标理解: {goal_understanding}
        可用工具: {tools}
        
        请返回子任务列表（JSON格式）：
        [
            {{
                "task_id": "task_1",
                "description": "任务描述",
                "required_tool": "需要的工具",
                "dependencies": ["task_0"],
                "estimated_duration": 120
            }}
        ]
        """
        return self.llm.ask(prompt, response_format="json")
    
    def _order_subtasks(self, subtasks: list) -> list:
        """确定执行顺序（拓扑排序）"""
        # 构建依赖图
        graph = self._build_dependency_graph(subtasks)
        
        # 拓扑排序
        return self._topological_sort(graph)

class Executor:
    """任务执行器"""
    
    def __init__(self):
        self.tool_registry = ToolRegistry()
        self.state_manager = StateManager()
        self.error_handler = ErrorHandler()
    
    def execute(self, plan: Plan) -> ExecutionResult:
        """执行计划"""
        results = {}
        
        for step in plan.steps:
            # 步骤1：检查依赖是否满足
            if not self._check_dependencies(step, results):
                self._handle_dependency_failure(step)
                continue
            
            # 步骤2：准备参数
            params = self._prepare_params(step, results)
            
            # 步骤3：执行工具
            try:
                # 记录执行开始
                self.state_manager.update(step.task_id, "running")
                
                # 调用工具
                tool = self.tool_registry.get_tool(step.required_tool)
                result = tool.execute(**params)
                
                # 记录执行成功
                self.state_manager.update(step.task_id, "completed")
                results[step.task_id] = result
                
            except Exception as e:
                # 错误处理
                self.state_manager.update(step.task_id, "failed")
                recovered = self.error_handler.recover(step, e, results)
                
                if recovered:
                    results[step.task_id] = recovered
                else:
                    raise  # 无法恢复，终止执行
        
        return ExecutionResult(results=results, status="completed")
    
    def _prepare_params(self, step: Step, previous_results: dict) -> dict:
        """准备执行参数"""
        # 从依赖的结果中提取参数
        params = {}
        for dep in step.dependencies:
            params.update(previous_results[dep])
        
        # 合合步骤的原始参数
        params.update(step.params)
        
        return params

class ToolRegistry:
    """工具注册中心"""
    
    def __init__(self):
        self.tools = {}
        self.tool_schemas = {}
    
    def register(self, tool: Tool):
        """注册工具"""
        self.tools[tool.name] = tool
        self.tool_schemas[tool.name] = tool.get_schema()
    
    def get_tool(self, name: str) -> Tool:
        """获取工具"""
        return self.tools.get(name)
    
    def get_all_schemas(self) -> dict:
        """获取所有工具的schema（用于LLM理解）"""
        return self.tool_schemas


class Tool:
    """工具基类"""
    
    def __init__(self, name: str, description: str: str):
        self.name = name
        self.description = description
    
    def execute(self, **kwargs) -> Any:
        """执行工具（子类实现）"""
        raise NotImplementedError
    
    def get_schema(self) -> dict:
        """返回工具的schema（用于LLM理解）"""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self._get_parameter_schema()
        }


# 具体工具示例
class WebSearchTool(Tool):
    """网络搜索工具"""
    
    def __init__(self, search_api):
        super().__init__(
            name="web_search",
            description="在互联网上搜索信息"
        )
        self.search_api = search_api
    
    def execute(self, query: str, num_results: int = 5) -> list:
        """执行搜索"""
        return self.search_api.search(query, num_results)
    
    def _get_parameter_schema(self) -> dict:
        """参数schema"""
        return {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                },
                "num_results": {
                    "type": "integer",
                    "description": "返回结果数量",
                    "default": 5
                }
            },
            "required": ["query"]
        }

class MemorySystem:
    """记忆系统"""
    
    def __init__(self):
        # 短期记忆（工作记忆）
        self.working_memory = []
        
        # 长期记忆（知识库）
        self.long_term_memory = VectorDatabase()
        
        # 语义记忆（概念知识）
        self.semantic_memory = KnowledgeBase()
        
        # 情景记忆（经验记录）
        self.episodic_memory = EventLogger()
    
    def store(self, content: str, metadata: dict = None):
        """存储记忆"""
        # 生成嵌入向量
        embedding = self._embed(content)
        
        # 存储到长期记忆
        memory_id = self.long_term_memory.store(
            content=content,
            embedding=embedding,
            metadata=metadata or {}
        )
        
        # 同时记录到情景记忆
        self.episodic_memory.log(
            event_id=memory_id,
            content=content,
            timestamp=datetime.now()
        )
        
        return memory_id
    
    def retrieve(self, query: str, top_k: int = 5) -> list:
        """检索相关记忆"""
        # 生成查询向量
        query_embedding = self._embed(query)
        
        # 向量检索
        results = self.long_term_memory.search(
            vector=query_embedding,
            top_k=top_k
        )
        
        return results
    
    def update(self, memory_id: str, new_content: str):
        """更新记忆"""
        # 先删除旧的
        self.delete(memory_id)
        
        # 再存储新的
        return self.store(new_content, {"updated_from": memory_id})
    
    def delete(self, memory_id: str):
        """删除记忆"""
        self.long_term_memory.delete(memory_id)

class MonitoringSystem:
    """监控与观测系统"""
    
    def __init__(self):
        self.metrics = MetricsCollector()
        self.tracer = Tracer()
        self.logger = StructuredLogger()
    
    def track_execution(self, func):
        """追踪执行过程"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 开始追踪
            trace_id = self.tracer.start_trace(func.__name__)
            
            try:
                # 记录开始时间
                start_time = time.time()
                
                # 执行函数
                result = func(*args, **kwargs)
                
                # 记录成功指标
                duration = time.time() - start_time
                self.metrics.record(
                    name=f"{func.__name__}.duration",
                    value=duration,
                    tags={"status": "success"}
                )
                
                return result
            
            except Exception as e:
                # 记录失败指标
                self.metrics.record(
                    name=f"{func.__name__}.errors",
                    value=1,
                    tags={"error_type": type(e).__name__}
                )
                
                # 记录错误日志
                self.logger.error(
                    trace_id=trace_id,
                    error=str(e),
                    stack_trace=traceback.format_exc()
                )
                
                raise
            
            finally:
                # 结束追踪
                self.tracer.end_trace(trace_id)
        
        return wrapper


class MetricsCollector:
    """指标收集器"""
    
    def __init__(self):
        self.counters = defaultdict(int)
        self.gauges = defaultdict(float)
        self.histograms = defaultdict(list)
    
    def record(self, name: str, value: float, tags: dict = None):
        """记录指标"""
        # 根据指标类型存储
        if name.endswith('.duration'):
            self.histograms[name].append(value)
        elif name.endswith('.errors'):
            self.counters[name] += int(value)
        else:
            self.gauges[name] = value
    
    def get_summary(self) -> dict:
        """获取指标摘要"""
        summary = {}
        
        # 计算直方图的统计值
        for name, values in self.histograms.items():
            if values:
                summary[f"{name}.avg"] = sum(values) / len(values)
                summary[f"{name}.max"] = max(values)
                summary[f"{name}.min"] = min(values)
        
        # 合并计数器和仪表
        summary.update(self.counters)
        summary.update(self.gauges)
        
        return summary

class SecurityLayer:
    """安全层"""
    
    def __init__(self):
        self.audit_logger = AuditLogger()
        self.permission_checker = PermissionChecker()
        self.content_filter = ContentFilter()
        self.rate_limiter = RateLimiter()
    
    def before_execution(self, user_id: str, action: Action) -> bool:
        """执行前检查"""
        
        # 检查1：权限验证
        if not self.permission_checker.check(user_id, action):
            self.audit_logger.log(
                event="permission_denied",
                user_id=user_id,
                action=action.name
            )
            return False
        
        # 检查2：内容过滤
        if hasattr(action, 'content') and self.content_filter.is_violation(action.content):
            self.audit_logger.log(
                event="content_violation",
                user_id=user_id,
                content=action.content
            )
            return False
        
        # 检查3：速率限制
        if not self.rate_limiter.allow(user_id, action.name):
            self.audit_logger.log(
                event="rate_limit_exceeded",
                user_id=user_id,
                action=action.name
            )
            return False
        
        return True
    
    def after_execution(self, user_id: str, action: Action, result: Any):
        """执行后审计"""
        self.audit_logger.log(
            event="action_completed",
            user_id=user_id,
            action=action.name,
            result_size=len(str(result))
        )


class PermissionChecker:
    """权限检查器"""
    
    def __init__(self):
        self.role_policies = {
            'admin': ['*'],  # 管理员拥有所有权限
            'user': ['read', 'write_own', 'search'],
            'guest': ['read', 'search']
        }
    
    def check(self, user_id: str, action: Action) -> bool:
        """检查用户是否有执行权限"""
        user_role = self._get_user_role(user_id)
        allowed_actions = self.role_policies.get(user_role, [])
        
        if '*' in allowed_actions:
            return True
        
        return action.name in allowed_actions