Firecracker在航空营销多智能体中的应用

from strands import Agent, tool
from firecracker_sandbox import eval_fn

@tool
def execute_python(code: str) -> str:
    """在Firecracker虚拟机中执行Python代码"""
    import json
    import re
    
    result, _ = eval_fn(code)
    
    # 尝试从输出中提取JSON
    try:
        # 查找JSON格式的输出
        json_pattern = r'\{[^{}]*\}'
        matches = re.findall(json_pattern, result)
        
        for match in matches:
            try:
                parsed_json = json.loads(match)
                # 如果成功解析JSON，返回格式化的结果
                return f"JSON Result: {json.dumps(parsed_json, indent=2)}\n\nRaw Output: {result}"
            except json.JSONDecodeError:
                continue
        
        # 如果没找到有效JSON，返回原始结果
        return f"Raw Output: {result}"
        
    except Exception as e:
        return f"Error processing output: {e}\nRaw Output: {result}"

@tool
def get_user_data(user_id="U001"):
    """从航空公司 CRM（客户关系管理）系统中提取指定用户的画像数据。是评估**"客户能出多少钱"和"客户有多大动力升舱"**的核心依据。
    Args:
        user_id: 用户ID，默认为"U001"
    Returns:
        统一格式的 JSON 响应，字段如下：
            name (str): 用户姓名，用于生成个性化话术。
            points (int): 用户账户内当前的可用里程积分。可以根据此数值决定"积分抵扣"的比例。
            sensitivity (float): 价格敏感度评分（0.0 到 1.0）。数值越高表示客户对价格越敏感，需要给出更激进的折扣。
            tier (str): 会员等级（如 Blue, Silver, Gold, Platinum）。等级越高，通常意味着客户的忠诚度越高，但对服务品质也更挑剔。
    """
    data = {
        "U001": {"name": "张先生", "points": 150000, "sensitivity": 0.2, "tier": "Gold"},
        "U002": {"name": "李女士", "points": 8000, "sensitivity": 0.8, "tier": "Silver"},
        "U003": {"name": "王医生", "points": 45000, "sensitivity": 0.5, "tier": "Blue"}
    }
    return data.get(user_id)

@tool
def get_flight_data(flight_id="NW123"):
    """从收益管理系统（RMS）中提取特定航班的实时运行状态。是评估**"这笔生意的成本是多少"以及"库存压力有多大"**的关键。
    Args:
        flight_id: 航班ID，默认为"NW123"
    Returns:
        统一格式的 JSON 响应，字段如下：
            seats_left (int): 目标舱位（即公务舱）当前的剩余空位数。空位越多，则越需要降价促销。
            hours_to_fly (int): 距离飞机起飞的剩余小时数。是判断"库存价值贬值速度"的关键变量。
            full_diff (float): 经济舱与目标舱位之间的标准现金差价（全价差）。是计算折扣的基准点。
            marginal_cost (float): 边际成本。指多承载一名升舱旅客所增加的实际成本（如餐食、清洁、燃油微增）。这是报价的绝对底线，现金报价不得低于此值。
    """
    # full_diff: 经济舱到公务舱的标准现金价差
    return {
        "NW123": {"seats_left": 5, "hours_to_fly": 12, "full_diff": 2000, "marginal_cost": 400},
        "CD567": {"seats_left": 2, "hours_to_fly": 72, "full_diff": 3500, "marginal_cost": 700}
    }

def main() -> None:
    """主程序入口"""
    system_prompt = """You are an AI airline revenue analyst helping customers find the best upgrade deals.

Your role:
1. Use get_user_data() and get_flight_data() to gather information
2. Use execute_python() to calculate optimal pricing strategies
3. Present results with CLEAR, SPECIFIC pricing information

CRITICAL: When calculating offers, ensure:
- Points used cannot exceed the customer's available points
- Cash price must be clearly stated
- Total cost breakdown must be accurate
- All calculations must be realistic and feasible

PYTHON CODE REQUIREMENTS:
- Write ALL Python code in English only (no Chinese comments, strings, or variable names)
- Use English variable names and comments only
- All strings in Python code must be in English
- Only your final response to the user should be in Chinese

MANDATORY: Every Python code block MUST end with a print() statement that outputs a JSON object with this exact format:
```python
import json
result = {
    "cash_price": 1200,
    "points_used": 8000, 
    "total_cost": 2000,
    "discount_percent": 60
}
print(json.dumps(result))
```

After getting the JSON result, present a customer-friendly upgrade offer in Chinese with:
- Exact cash amount to pay (yuan)
- Exact points to use (must be ≤ available points)  
- Clear total cost breakdown
- What the customer gets"""
    
    agent = Agent(
        tools=[execute_python, get_user_data, get_flight_data],
        system_prompt=system_prompt
    )
    
    task = """# Role
你是一名航空公司 AI 收益精算师。你擅长使用 Python 执行 CodeAct 逻辑，通过实时计算为客户提供"最难拒绝"的升舱方案。

# Context
客户正在查看其预订的机票。你需要通过 API 获取客户画像和航班实时库存，然后编写并运行 Python 代码，计算出一个能平衡"航司收益最大化"与"客户转化率最大化"的升舱报价。

# Constraints & Logic
1. **收益模型**：期望收益 $E = P(成交) \times (现金收入 + 积分财务价值)$。
2. **定价原则**：
   - 离起飞越近（<24h）且空位越多，现金价格越低。
   - 客户积分余额越高，应引导其消耗更多积分（积分价值按 0.05元/分折算）。
   - 报价不得低于目标舱位边际成本（假设为原差价的 20%）。
3. **CodeAct 要求**：
   - 必须编写 Python 脚本进行方案模拟。
   - 对比 [4折, 6折, 8折] 等不同档位，选择期望收益最高的一组。

# Output Format
1. 思考过程（分析库存与客户画像）。
2. 生成并执行 Python 代码块。
3. 给出最终给客户展示的文案。

# User Query
我的用户ID是U002，NW123航班升舱太贵了，有没有便宜点的方案？
    """
    result = agent(task)
    print(f"\n完整回复:\n{result}")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Firecracker沙箱 - 安全的Python代码执行环境
提供基于Firecracker微虚拟机的Python代码执行服务，确保代码在隔离环境中安全运行。
"""

import json
import time
import pexpect
import requests_unixsocket
import os
import atexit
from typing import Tuple, Dict, Any, Optional

FIRECRACKER_SOCKET = "/tmp/firecracker.sock"
DEFAULT_KERNEL = "/home/ec2-user/codeact/llm-vmlinux-l.bin"
DEFAULT_ROOTFS = "/home/ec2-user/codeact/llm-rootfs-1.ext4"

firecracker_process: Optional[pexpect.spawn] = None
vm_console: Optional[pexpect.spawn] = None

def firecracker_api_put(path: str, data: dict) -> dict:
    """向Firecracker API发送PUT请求"""
    session = requests_unixsocket.Session()
    url = f"http+unix://{FIRECRACKER_SOCKET.replace('/', '%2F')}{path}"
    headers = {"Content-Type": "application/json"}
    
    resp = session.put(url, headers=headers, data=json.dumps(data), timeout=5)
    if not resp.ok:
        raise RuntimeError(f"Firecracker API PUT {path} failed: {resp.text}")
    
    return resp.json() if resp.text.strip() else {}

def cleanup_firecracker() -> None:
    """清理Firecracker进程和相关资源"""
    global firecracker_process, vm_console
    
    if vm_console:
        vm_console.close()
        vm_console = None
        
    if firecracker_process:
        firecracker_process.terminate()
        firecracker_process.wait()
        firecracker_process = None
        
    if os.path.exists(FIRECRACKER_SOCKET):
        os.unlink(FIRECRACKER_SOCKET)

def start_vm(kernel: str = DEFAULT_KERNEL, rootfs: str = DEFAULT_ROOTFS, buffer_size: int = 65536) -> None:
    """启动Firecracker虚拟机"""
    global vm_console, firecracker_process
    
    if not os.path.exists(kernel) or not os.path.exists(rootfs):
        raise FileNotFoundError("Kernel or rootfs not found")
    
    if os.path.exists(FIRECRACKER_SOCKET):
        os.unlink(FIRECRACKER_SOCKET)
    
    # 启动Firecracker进程，使用动态缓冲区
    firecracker_process = pexpect.spawn(
        '/usr/local/bin/firecracker', 
        ['--api-sock', FIRECRACKER_SOCKET],
        encoding='utf-8', 
        timeout=30, 
        maxread=buffer_size, 
        searchwindowsize=min(buffer_size // 8, 8192)
    )
    vm_console = firecracker_process
    atexit.register(cleanup_firecracker)
    
    for _ in range(10):
        if os.path.exists(FIRECRACKER_SOCKET):
            break
        time.sleep(0.5)
    else:
        raise RuntimeError("Firecracker socket not created")

    # 配置VM
    firecracker_api_put("/boot-source", {
        "kernel_image_path": kernel,
        "boot_args": "console=ttyS0 reboot=k panic=1 pci=off"
    })
    
    firecracker_api_put("/drives/rootfs", {
        "drive_id": "rootfs",
        "path_on_host": rootfs,
        "is_root_device": True,
        "is_read_only": False
    })
    
    firecracker_api_put("/actions", {"action_type": "InstanceStart"})
    
    vm_console.expect('login:', timeout=15)
    vm_console.sendline('root')
    
    idx = vm_console.expect(['Password:', '#'], timeout=5)
    if idx == 0:
        vm_console.sendline('root')
        vm_console.expect('#', timeout=5)

def convert_to_oneline(script: str) -> str:
    """将多行Python脚本转换为单行执行格式"""
    lines = [line for line in script.split('\n') 
             if line.strip() and not line.strip().startswith('#')]
    
    code_block = '\\n'.join(lines)
    return f'exec("""{code_block}""")'

def extract_json_from_output(output: str) -> Optional[str]:
    """从VM输出中提取JSON数据"""
    import re
    json_patterns = [
        r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}',  # 简单嵌套
        r'\{(?:[^{}]|(?:\{[^{}]*\}))*\}',    # 递归嵌套
        r'\{.*?\}',                          # 贪婪匹配
    ]
    
    for pattern in json_patterns:
        json_matches = re.findall(pattern, output, re.DOTALL)
        
        for match in json_matches:
            try:
                json.loads(match)  # 验证JSON有效性
                return match
            except json.JSONDecodeError:
                continue
    
    return None

def clean_vm_output(output: str) -> str:
    """清理VM输出，移除控制字符和命令回显"""
    lines = output.split('\r\n')
    filtered = []
    
    for line in lines:
        line = line.strip()
        if (line and 
            not line.startswith('python3') and
            not line.startswith('[') and  # 内核消息
            not line.endswith('#') and
            'localhost' not in line and
            '\x1b' not in line):  # ANSI转义序列
            filtered.append(line)
    
    return '\n'.join(filtered)

def calculate_buffer_size(script: str) -> int:
    """根据代码大小计算所需的缓冲区大小"""
    base_size = 32768  # 32KB
    code_length = len(script)
    estimated_output = code_length * 3
    buffer_size = max(base_size, min(estimated_output + base_size, 1048576))
    return buffer_size

def eval_fn(script: str, context: Optional[dict] = None) -> Tuple[str, Dict[str, Any]]:
    """在Firecracker VM中执行Python代码"""
    global vm_console
    
    print("-" * 40)
    print(f"LLM-Generated Python Code: \n{script}")
    print("~" * 40)
    
    buffer_size = calculate_buffer_size(script)
    if vm_console is None:
        start_vm(buffer_size=buffer_size)
    else:
        current_maxread = getattr(vm_console, 'maxread', 65536)
        if buffer_size > current_maxread:
            cleanup_firecracker()
            start_vm(buffer_size=buffer_size)
    
    try:
        oneline_script = convert_to_oneline(script)
        escaped_script = oneline_script.replace('"', '\\"').replace('$', '\\$')
        vm_console.sendline(f'python3 -c "{escaped_script}"')
        vm_console.expect('#', timeout=60)
        output = vm_console.before.strip()
        
        print("-" * 40)
        print(f"VM raw output ({len(output)} chars): {output}")
        print("=" * 40)
        
        json_result = extract_json_from_output(output)
        if json_result:
            return json_result, {}
        cleaned_output = clean_vm_output(output)
        return cleaned_output, {}
    except pexpect.TIMEOUT:
        return "Error: Execution timeout", {}
    except Exception as e:
        return f"Error: {str(e)}", {}