Codex智能体实战：从零系统构建企业级智能体应用

┌─────────────────────────────────────────────────────┐
│                   智能体系统架构                      │
│                                                     │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌────────┐│
│  │ 感知模块 │→│ 推理引擎 │→│ 规划模块 │→│ 执行器 ││
│  │(Perception)│(Reasoning)│(Planning)│(Executor)││
│  └─────────┘  └─────────┘  └─────────┘  └────────┘│
│       │            │            │           │       │
│       └────────────┴────────────┴───────────┘       │
│                         │                           │
│                    ┌────▼────┐                      │
│                    │ 记忆系统 │                      │
│                    │(Memory) │                      │
│                    └─────────┘                      │
└─────────────────────────────────────────────────────┘

用户输入 → LLM理解 → 生成Python代码 → 沙箱执行 → 返回结果

# 基础依赖
pip install openai langchain langchain-community
pip install e2b-code-interpreter  # 代码执行沙箱
pip install qdrant-client  # 向量数据库
pip install pandas numpy matplotlib  # 数据处理
pip install fastapi uvicorn  # API服务

FROM python:3.11-slim

# 限制系统调用
RUN apt-get update && apt-get install -y \
    libseccomp2 \
    && rm -rf /var/lib/apt/lists/*

# 安装常用库
RUN pip install numpy pandas requests beautifulsoup4

# 设置资源限制
# 运行时通过docker run --memory=512m --cpus=0.5 控制

WORKDIR /workspace
CMD ["python", "-m", "http.server", "8000"]

import os
from openai import OpenAI
from langchain.agents import AgentExecutor
from langchain_openai import ChatOpenAI

# Codex API配置（通过OpenAI接口调用）
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url="https://api.openai.com/v1",  # Codex模型路径
)

# 使用LangChain封装
llm = ChatOpenAI(
    model="gpt-4-codex",  # 具体模型名
    temperature=0.1,  # 代码生成需要低温度
    max_tokens=4096,
    api_key=os.getenv("OPENAI_API_KEY"),
)

# 基础代码生成测试
def generate_code(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "你是一个Python专家。只输出代码，不要解释。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.1,
    )
    return response.choices[0].message.content

# 测试
code = generate_code("写一个函数，计算列表中所有偶数的平方和")
print(code)

┌─────────────────────────────────────────────────────────────┐
│                      应用层（API / Webhook）               │
├─────────────────────────────────────────────────────────────┤
│                   智能体编排层（Orchestrator）              │
│        ┌────────────┐  ┌────────────┐  ┌────────────┐     │
│        │  Planner   │  │  Executor  │  │  Reflector │     │
│        └────────────┘  └────────────┘  └────────────┘     │
├─────────────────────────────────────────────────────────────┤
│                    工具层（Tools Registry）                 │
│   ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐          │
│   │DB查询│ │API调用│ │计算  │ │文件  │ │邮件  │          │
│   └──────┘ └──────┘ └──────┘ └──────┘ └──────┘          │
├─────────────────────────────────────────────────────────────┤
│                    记忆层（Memory）                         │
│   ┌────────────┐  ┌────────────┐  ┌────────────┐          │
│   │短期记忆    │  │长期记忆    │  │工作记忆    │          │
│   │(Redis)     │  │(向量库)    │  │(上下文)    │          │
│   └────────────┘  └────────────┘  └────────────┘          │
├─────────────────────────────────────────────────────────────┤
│                   基础设施层                                │
│      Codex引擎   │  沙箱执行器  │  日志/监控              │
└─────────────────────────────────────────────────────────────┘

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import json
import asyncio

class AgentState(Enum):
    IDLE = "idle"
    PLANNING = "planning"
    EXECUTING = "executing"
    REFLECTING = "reflecting"
    DONE = "done"
    ERROR = "error"

@dataclass
class AgentContext:
    """智能体上下文 - 在一次任务生命周期内传递"""
    task_id: str
    user_id: str
    input: str
    plan: List[Dict] = field(default_factory=list)
    current_step: int = 0
    memory: Dict[str, Any] = field(default_factory=dict)
    execution_history: List[Dict] = field(default_factory=list)
    max_iterations: int = 10

class SupplyChainAgent:
    """供应链调度智能体"""
    
    def __init__(self, codex_llm, tools_registry, memory_service, sandbox):
        self.llm = codex_llm
        self.tools = tools_registry
        self.memory = memory_service
        self.sandbox = sandbox
        self.max_retries = 3
    
    async def run(self, user_input: str, context: Optional[AgentContext] = None) -> Dict:
        """主入口：处理用户请求"""
        if context is None:
            context = AgentContext(
                task_id=self._generate_task_id(),
                user_id="default",
                input=user_input
            )
        
        try:
            # 1. 理解意图
            await self._understand_intent(context)
            
            # 2. 生成计划
            await self._plan(context)
            
            # 3. 执行计划（循环）
            while context.current_step < len(context.plan):
                step = context.plan[context.current_step]
                result = await self._execute_step(step, context)
                
                # 4. 反思：评估执行结果，决定下一步
                should_continue = await self._reflect(result, context)
                if not should_continue:
                    break
                
                context.current_step += 1
            
            # 5. 汇总结果
            return self._synthesize(context)
            
        except Exception as e:
            return {
                "status": "error",
                "error": str(e),
                "partial_results": context.execution_history
            }

async def _understand_intent(self, context: AgentContext):
    """使用Codex分析用户意图，不生成执行代码，只做分析"""
    
    prompt = f"""
你是一个供应链调度专家。分析以下用户请求，提取关键信息。

用户请求：{context.input}

请以JSON格式输出：
{{
    "intent": "查询/分析/决策/预警/报告",
    "entities": {{
        "time_range": "提取时间范围，如最近7天",
        "locations": ["涉及的地点"],
        "order_ids": ["涉及的订单号"],
        "metrics": ["涉及的指标"]
    }},
    "complexity": "simple/medium/complex",
    "suggested_approach": "简要描述你认为最合适的处理方式"
}}
"""
    
    response = await self.llm.apredict(prompt)
    try:
        analysis = json.loads(response)
        context.memory["intent_analysis"] = analysis
    except json.JSONDecodeError:
        # 降级处理
        context.memory["intent_analysis"] = {
            "intent": "unknown",
            "entities": {},
            "complexity": "medium"
        }

async def _plan(self, context: AgentContext):
    """生成执行计划 - 这是Codex智能体的核心"""
    
    # 获取可用工具列表
    tools_desc = self.tools.get_description()
    
    # 获取历史相关记忆
    relevant_memory = await self.memory.retrieve(
        context.input, 
        top_k=5
    )
    
    plan_prompt = f"""
你是一个供应链调度智能体。你需要为以下任务生成一个详细的执行计划。

任务描述：{context.input}

意图分析：{json.dumps(context.memory.get('intent_analysis', {}), ensure_ascii=False)}

可用工具：
{tools_desc}

相关历史经验：
{relevant_memory}

请生成一个执行计划，以JSON数组形式输出，每个步骤包含：
- step_id: 步骤编号
- action: 要执行的动作（调用工具/计算/查询等）
- tool: 使用的工具名称（如果是工具调用）
- code: 需要执行的Python代码（如果有）
- params: 参数
- depends_on: 依赖的步骤ID列表
- fallback: 失败时的备选方案

只输出JSON数组，不要其他内容。
"""
    
    response = await self.llm.apredict(plan_prompt)
    try:
        plan = json.loads(response)
        context.plan = plan
        context.memory["plan"] = plan
    except json.JSONDecodeError as e:
        # 如果JSON解析失败，尝试从响应中提取代码块
        plan = self._extract_plan_from_text(response)
        context.plan = plan

import docker
import tempfile
import subprocess
import time

class CodeExecutor:
    """安全的代码执行器"""
    
    def __init__(self, timeout=30, memory_limit="512m", cpu_limit="0.5"):
        self.timeout = timeout
        self.memory_limit = memory_limit
        self.cpu_limit = cpu_limit
        self.client = docker.from_env()
    
    def execute(self, code: str, context: Dict) -> Dict:
        """
        在Docker沙箱中执行代码
        返回: { "output": ..., "error": ..., "execution_time": ... }
        """
        start_time = time.time()
        
        # 构建完整脚本（注入上下文变量）
        full_code = self._prepare_script(code, context)
        
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(full_code)
            script_path = f.name
        
        try:
            # 运行Docker容器
            container = self.client.containers.run(
                image="codex-sandbox:latest",
                command=f"python {script_path}",
                volumes={os.path.dirname(script_path): {'bind': '/workspace', 'mode': 'ro'}},
                mem_limit=self.memory_limit,
                nano_cpus=int(float(self.cpu_limit) * 1e9),
                network_disabled=False,  # 允许网络请求
                detach=True,
                auto_remove=False,
            )
            
            # 等待完成（带超时）
            result = container.wait(timeout=self.timeout)
            
            # 获取日志
            logs = container.logs(stdout=True, stderr=True).decode('utf-8')
            
            # 清理容器
            container.remove()
            
            execution_time = time.time() - start_time
            
            return {
                "success": result["StatusCode"] == 0,
                "output": logs if result["StatusCode"] == 0 else "",
                "error": logs if result["StatusCode"] != 0 else "",
                "execution_time": execution_time,
            }
            
        except docker.errors.APIError as e:
            return {
                "success": False,
                "error": f"Docker error: {str(e)}",
                "execution_time": time.time() - start_time,
            }
        except Exception as e:
            return {
                "success": False,
                "error": f"Execution error: {str(e)}",
                "execution_time": time.time() - start_time,
            }
        finally:
            if os.path.exists(script_path):
                os.unlink(script_path)
    
    def _prepare_script(self, code: str, context: Dict) -> str:
        """为脚本注入上下文变量和辅助函数"""
        return f'''
import json
import sys
from typing import Dict, List, Any

# 注入上下文
context = {json.dumps(context, default=str)}

# 辅助函数：安全打印
def safe_print(*args, **kwargs):
    try:
        print(*args, **kwargs)
    except:
        pass

# 辅助函数：结果保存
def save_result(data):
    with open("/workspace/result.json", "w") as f:
        json.dump(data, f, default=str)

# 用户代码
{code}
'''

async def _reflect(self, result: Dict, context: AgentContext) -> bool:
    """反思执行结果，决定是否继续"""
    
    current_step = context.plan[context.current_step]
    
    reflection_prompt = f"""
你是一个供应链调度智能体，正在执行一个任务。

当前步骤：{json.dumps(current_step, ensure_ascii=False)}
执行结果：{json.dumps(result, ensure_ascii=False)}
已完成的步骤数量：{context.current_step}
总步骤数：{len(context.plan)}

请评估：
1. 当前步骤是否成功执行？
2. 结果是否符合预期？
3. 是否有异常需要处理？
4. 是否需要调整后续计划？
5. 是否应该继续执行下一步？

以JSON格式输出：
{{
    "success": true/false,
    "confidence": 0.0-1.0,
    "should_continue": true/false,
    "adjustment_needed": true/false,
    "adjusted_plan": null 或 调整后的计划步骤列表,
    "notes": "简要说明"
}}
"""
    
    response = await self.llm.apredict(reflection_prompt)
    try:
        reflection = json.loads(response)
        
        # 如果需要调整计划，更新context.plan
        if reflection.get("adjustment_needed") and reflection.get("adjusted_plan"):
            context.plan = reflection["adjusted_plan"]
            context.memory["plan_adjusted"] = True
        
        # 记录反思结果
        context.execution_history.append({
            "step": context.current_step,
            "result": result,
            "reflection": reflection
        })
        
        return reflection.get("should_continue", False)
        
    except json.JSONDecodeError:
        # 默认继续
        return True

from typing import Callable, Dict, Any, Optional
import inspect

class ToolRegistry:
    """工具注册与调度中心"""
    
    def __init__(self):
        self._tools: Dict[str, Dict] = {}
    
    def register(self, name: str, description: str, func: Callable, 
                 schema: Optional[Dict] = None):
        """注册一个工具"""
        self._tools[name] = {
            "name": name,
            "description": description,
            "func": func,
            "schema": schema or self._infer_schema(func),
            "usage_count": 0,
            "total_time": 0.0,
            "error_count": 0,
        }
    
    def get(self, name: str) -> Optional[Callable]:
        tool = self._tools.get(name)
        return tool["func"] if tool else None
    
    def get_description(self) -> str:
        """生成工具描述，供LLM理解"""
        desc = "可用工具列表：\n"
        for name, tool in self._tools.items():
            desc += f"\n- {name}: {tool['description']}"
            desc += f"\n  参数: {json.dumps(tool['schema'], ensure_ascii=False)}"
        return desc
    
    async def call(self, name: str, **kwargs) -> Any:
        """调用工具（带监控）"""
        tool = self._tools.get(name)
        if not tool:
            raise ValueError(f"Tool {name} not found")
        
        start = time.time()
        try:
            result = tool["func"](**kwargs)
            if asyncio.iscoroutine(result):
                result = await result
            tool["usage_count"] += 1
            return result
        except Exception as e:
            tool["error_count"] += 1
            raise
        finally:
            tool["total_time"] += time.time() - start
    
    def _infer_schema(self, func: Callable) -> Dict:
        """从函数签名推断参数schema"""
        sig = inspect.signature(func)
        schema = {
            "type": "object",
            "properties": {},
            "required": []
        }
        for name, param in sig.parameters.items():
            if name in ('self', 'cls', 'context'):
                continue
            schema["properties"][name] = {
                "type": self._type_to_json(param.annotation)
            }
            if param.default == inspect.Parameter.empty:
                schema["required"].append(name)
        return schema
    
    @staticmethod
    def _type_to_json(annotation) -> str:
        if annotation == str:
            return "string"
        if annotation == int:
            return "integer"
        if annotation == float:
            return "number"
        if annotation == bool:
            return "boolean"
        if annotation == list:
            return "array"
        if annotation == dict:
            return "object"
        return "string"

# 示例工具1：订单查询
class OrderTools:
    @staticmethod
    def get_order(order_id: str) -> Dict:
        """查询订单详情"""
        # 实际调用TMS系统API
        return {
            "order_id": order_id,
            "status": "in_transit",
            "current_location": "上海港",
            "destination": "洛杉矶",
            "eta": "2025-07-20",
            "cargo": [
                {"sku": "ELEC-001", "quantity": 1000, "weight_kg": 2500}
            ]
        }
    
    @staticmethod
    def list_orders(day_range: int = 7, status: Optional[str] = None) -> List[Dict]:
        """列出指定天数内的订单"""
        # 查询TMS订单列表
        return [...]

# 示例工具2：路线优化
class RouteTools:
    @staticmethod
    def find_alternative_routes(origin: str, destination: str, 
                                avoid_locations: List[str] = None) -> List[Dict]:
        """寻找替代运输路线"""
        # 调用地图API + 内部路线库
        return [
            {"route_id": "R001", "path": ["上海", "宁波", "洛杉矶"], 
             "duration_days": 25, "cost": 85000},
            {"route_id": "R002", "path": ["上海", "深圳", "洛杉矶"], 
             "duration_days": 28, "cost": 72000},
        ]
    
    @staticmethod
    def calculate_cost(route_id: str, cargo_weight: float) -> float:
        """计算特定路线的成本"""
        # 基于历史数据和当前运费率计算
        return 0.0

# 示例工具3：预警分析
class AlertTools:
    @staticmethod
    def get_weather_alerts(location: str, days_ahead: int = 7) -> List[Dict]:
        """获取指定位置的天气预警"""
        # 调用天气API
        return [
            {"type": "台风", "level": "橙色", "location": "上海", 
             "start_time": "2025-07-15", "end_time": "2025-07-17"}
        ]
    
    @staticmethod
    def evaluate_impact(alert: Dict, orders: List[Dict]) -> Dict:
        """评估预警对订单的影响"""
        affected_orders = []
        for order in orders:
            if order["current_location"] == alert["location"]:
                affected_orders.append(order)
        return {
            "affected_count": len(affected_orders),
            "affected_orders": affected_orders,
            "severity": alert["level"],
            "recommended_action": "redirect"
        }

# 注册所有工具
tools_registry = ToolRegistry()
tools_registry.register(
    "get_order", 
    "根据订单ID查询订单详情", 
    OrderTools.get_order
)
tools_registry.register(
    "list_orders", 
    "列出最近N天的所有订单", 
    OrderTools.list_orders
)
tools_registry.register(
    "find_alternative_routes", 
    "查找从起点到终点的替代运输路线", 
    RouteTools.find_alternative_routes
)
tools_registry.register(
    "get_weather_alerts", 
    "获取指定位置的天气预警信息", 
    AlertTools.get_weather_alerts
)
tools_registry.register(
    "evaluate_impact", 
    "评估天气预警对订单的影响", 
    AlertTools.evaluate_impact
)

async def _execute_step(self, step: Dict, context: AgentContext) -> Dict:
    """执行单个计划步骤"""
    
    action = step.get("action")
    
    if action == "call_tool":
        # 工具调用模式
        tool_name = step.get("tool")
        params = step.get("params", {})
        
        # 解析参数中的动态引用（如从上下文中取值）
        resolved_params = self._resolve_params(params, context)
        
        try:
            result = await self.tools.call(tool_name, **resolved_params)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    elif action == "execute_code":
        # 代码执行模式（Codex核心）
        code = step.get("code", "")
        
        # 补充上下文变量到代码中
        enriched_code = self._enrich_code(code, context)
        
        # 在沙箱中执行
        result = self.sandbox.execute(enriched_code, context.memory)
        
        if result["success"]:
            # 尝试解析输出为结构化数据
            try:
                output = json.loads(result["output"])
                return {"success": True, "result": output}
            except:
                return {"success": True, "result": result["output"]}
        else:
            return {"success": False, "error": result["error"]}
    
    else:
        return {"success": False, "error": f"Unknown action: {action}"}

from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, Distance, VectorParams
import numpy as np
from sentence_transformers import SentenceTransformer

class VectorMemory:
    """基于向量的长期记忆系统"""
    
    def __init__(self, collection_name: str = "agent_memory", 
                 embedding_model: str = "all-MiniLM-L6-v2"):
        self.client = QdrantClient(host="localhost", port=6333)
        self.collection = collection_name
        self.encoder = SentenceTransformer(embedding_model)
        
        # 确保集合存在
        self._ensure_collection()
    
    def _ensure_collection(self):
        collections = self.client.get_collections().collections
        if not any(c.name == self.collection for c in collections):
            self.client.create_collection(
                collection_name=self.collection,
                vectors_config=VectorParams(
                    size=384,  # all-MiniLM-L6-v2的维度
                    distance=Distance.COSINE
                )
            )
    
    def store(self, text: str, metadata: Dict):
        """存储一条记忆"""
        vector = self.encoder.encode(text).tolist()
        point_id = self._generate_id()
        
        self.client.upsert(
            collection_name=self.collection,
            points=[PointStruct(
                id=point_id,
                vector=vector,
                payload={
                    "text": text,
                    "metadata": metadata,
                    "timestamp": time.time()
                }
            )]
        )
        return point_id
    
    def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
        """检索相似记忆"""
        query_vector = self.encoder.encode(query).tolist()
        
        results = self.client.search(
            collection_name=self.collection,
            query_vector=query_vector,
            limit=top_k,
            with_payload=True
        )
        
        return [
            {
                "text": hit.payload["text"],
                "metadata": hit.payload["metadata"],
                "score": hit.score,
                "timestamp": hit.payload["timestamp"]
            }
            for hit in results
        ]
    
    def prune_old_memories(self, days_threshold: int = 30):
        """清理过期记忆"""
        cutoff = time.time() - days_threshold * 24 * 3600
        # 获取所有点，过滤掉过期的
        # （实际生产环境可以设置TTL）
        pass
    
    def _generate_id(self) -> int:
        return int(time.time() * 1000 + np.random.randint(0, 1000))

class WorkingMemory:
    """工作记忆 - 管理当前任务的临时信息"""
    
    def __init__(self, max_items: int = 20):
        self.max_items = max_items
        self._items: Dict[str, Any] = {}
        self._access_log: List[str] = []  # LRU跟踪
    
    def set(self, key: str, value: Any):
        """设置工作记忆项"""
        if len(self._items) >= self.max_items:
            # 移除最久未访问的项
            oldest = self._access_log.pop(0)
            del self._items[oldest]
        
        self._items[key] = value
        self._access_log.append(key)
    
    def get(self, key: str, default=None) -> Any:
        """获取工作记忆项（会更新访问时间）"""
        if key in self._items:
            self._access_log.remove(key)
            self._access_log.append(key)
        return self._items.get(key, default)
    
    def get_all(self) -> Dict:
        return self._items.copy()
    
    def clear(self):
        self._items.clear()
        self._access_log.clear()

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional
import uuid

app = FastAPI(title="Codex Supply Chain Agent")

class AgentRequest(BaseModel):
    query: str
    user_id: str
    session_id: Optional[str] = None
    context: Optional[Dict] = None
    async_mode: bool = False

class AgentResponse(BaseModel):
    task_id: str
    status: str
    result: Optional[Any] = None
    plan: Optional[List] = None
    execution_history: Optional[List] = None

# 智能体实例
agent = SupplyChainAgent(
    codex_llm=llm,
    tools_registry=tools_registry,
    memory_service=vector_memory,
    sandbox=code_executor
)

# 任务存储（生产环境用Redis）
task_store = {}

@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(request: AgentRequest, background_tasks: BackgroundTasks):
    """同步执行智能体任务"""
    task_id = str(uuid.uuid4())
    
    context = AgentContext(
        task_id=task_id,
        user_id=request.user_id,
        input=request.query,
    )
    
    if request.context:
        context.memory.update(request.context)
    
    try:
        result = await agent.run(request.query, context)
        return AgentResponse(
            task_id=task_id,
            status=result.get("status", "completed"),
            result=result.get("result"),
            plan=context.plan,
            execution_history=context.execution_history
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/agent/run_async")
async def run_agent_async(request: AgentRequest, background_tasks: BackgroundTasks):
    """异步执行智能体任务（适合长时间任务）"""
    task_id = str(uuid.uuid4())
    
    context = AgentContext(
        task_id=task_id,
        user_id=request.user_id,
        input=request.query,
    )
    
    # 存储初始状态
    task_store[task_id] = {"status": "pending", "result": None}
    
    async def execute():
        try:
            result = await agent.run(request.query, context)
            task_store[task_id] = {
                "status": "completed",
                "result": result,
                "plan": context.plan,
                "history": context.execution_history
            }
        except Exception as e:
            task_store[task_id] = {"status": "failed", "error": str(e)}
    
    background_tasks.add_task(execute)
    
    return {"task_id": task_id, "status": "pending"}

@app.get("/agent/task/{task_id}")
async def get_task_status(task_id: str):
    """查询异步任务状态"""
    task = task_store.get(task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

# 健康检查
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import Response

# 指标定义
agent_requests = Counter('agent_requests_total', 'Total agent requests', ['type'])
agent_duration = Histogram('agent_duration_seconds', 'Agent execution time', 
                           buckets=[1, 5, 10, 30, 60, 120])
agent_errors = Counter('agent_errors_total', 'Total agent errors', ['type'])
tool_calls = Counter('tool_calls_total', 'Total tool calls', ['tool_name'])
tool_duration = Histogram('tool_duration_seconds', 'Tool execution time', 
                          ['tool_name'])

@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type="text/plain")

class RobustAgent(SupplyChainAgent):
    """带容错能力的智能体包装器"""
    
    async def run(self, user_input: str, context: Optional[AgentContext] = None):
        try:
            # 主逻辑
            return await super().run(user_input, context)
        except TimeoutError:
            # 超时降级：返回部分结果 + 请求人工介入
            return {
                "status": "partial",
                "message": "任务处理超时，已生成部分结果",
                "partial_results": context.execution_history if context else [],
                "needs_human": True
            }
        except CodeExecutionError as e:
            # 代码执行失败：尝试用规则引擎兜底
            return await self._fallback_rule_engine(user_input, context)
        except Exception as e:
            # 未知错误：记录并上报
            logger.error(f"Unexpected error: {e}", exc_info=True)
            return {
                "status": "error",
                "message": "系统遇到意外错误，已通知工程师处理",
                "error_id": str(uuid.uuid4())
            }
    
    async def _fallback_rule_engine(self, user_input: str, context: Optional[AgentContext]):
        """规则引擎降级"""
        # 基于关键词的简单规则匹配
        if "订单" in user_input and "查询" in user_input:
            return {
                "status": "completed",
                "result": {"message": "使用规则引擎查询订单", "orders": []},
                "mode": "fallback_rule_engine"
            }
        return {
            "status": "failed",
            "message": "无法处理该请求，已转人工",
            "needs_human": True
        }

输入: "台风预警：上海港将于7月16日-17日关闭。请评估影响并给出建议。"

{
  "intent": "预警响应",
  "entities": {
    "event_type": "台风",
    "location": "上海港",
    "time_range": "2025-07-16 到 2025-07-17"
  },
  "plan": [
    {"step_id": 1, "action": "call_tool", "tool": "get_weather_alerts", 
     "params": {"location": "上海港", "days_ahead": 7}},
    {"step_id": 2, "action": "call_tool", "tool": "list_orders", 
     "params": {"day_range": 7}},
    {"step_id": 3, "action": "execute_code", 
     "code": "筛选出当前在上海港且预计16-17日出发/到达的订单"},
    {"step_id": 4, "action": "call_tool", "tool": "find_alternative_routes",
     "params": {"origin": "上海港", "destination": "洛杉矶", 
                "avoid_locations": ["上海港"]}},
    {"step_id": 5, "action": "execute_code",
     "code": "计算各替代方案的额外成本和延误时间"}
  ]
}

# 步骤3：筛选受影响订单
import json
from datetime import datetime, timedelta

def filter_affected_orders(orders, alert):
    """筛选受台风影响的订单"""
    affected = []
    alert_start = datetime.strptime(alert["start_time"], "%Y-%m-%d")
    alert_end = datetime.strptime(alert["end_time"], "%Y-%m-%d")
    
    for order in orders:
        if order["current_location"] != "上海港":
            continue
        
        # 检查订单的预计时间是否在预警期内
        eta = datetime.strptime(order["eta"], "%Y-%m-%d")
        if alert_start <= eta <= alert_end:
            affected.append(order)
        # 也检查是否在预警期内到达上海港
        if order.get("next_stop") == "上海港":
            next_stop_time = datetime.strptime(order.get("next_stop_time", "2025-07-17"), "%Y-%m-%d")
            if alert_start <= next_stop_time <= alert_end:
                affected.append(order)
    
    return affected

affected = filter_affected_orders(orders, alert)
print(json.dumps({
    "affected_count": len(affected),
    "affected_orders": [o["order_id"] for o in affected],
    "total_orders_examined": len(orders)
}))

{
  "status": "completed",
  "result": {
    "summary": "台风预警影响评估完成",
    "affected_orders": 23,
    "impact": "高",
    "alternatives": [
      {"route": "上海→深圳→洛杉矶", "extra_days": 3, "extra_cost": 12000},
      {"route": "上海→宁波→洛杉矶", "extra_days": 2, "extra_cost": 8500}
    ],
    "recommendation": "建议立即启动备选方案2（经宁波），23个受影响订单中15个可改道，其余8个建议延迟发运",
    "details": "23个受影响订单中，15个为出口订单（目的地：洛杉矶），8个为进口订单（来源地：新加坡）"
  },
  "mode": "codex_agent",
  "confidence": 0.85
}