进程（Web 服务器）
├── 线程1：处理用户A的请求
├── 线程2：处理用户B的请求
├── 线程3：处理用户C的请求
└── ...（更多线程）
    │
    ├── 共享资源：数据库连接池、缓存
    ├── 独立资源：每个线程自己的栈、局部变量
    └── 潜在问题：多个线程同时修改同一个变量？

# 操作系统视角
class Process:
    def __init__(self):
        self.memory = Memory()  # 独立地址空间
        self.files = FileHandles()
        self.threads = []

# Agent 视角
class Agent:
    def __init__(self, model, tools):
        self.memory = Memory()      # Agent 的"内存"
        self.context = []          # Agent 的"上下文"
        self.sub_agents = []      # Sub-Agent 们

时间线：
T1: Agent A 读取 shared_memory["result"] = None
T2: Agent A 开始计算
T3: Agent B 读取 shared_memory["result"] = None  （还是空的！）
T4: Agent A 计算完成，写入 shared_memory["result"] = "答案A"
T5: Agent B 计算完成，写入 shared_memory["result"] = "答案B"  （覆盖了！）

场景：
- Agent A 等 Agent B 的结果
- Agent B 等 Agent A 的结果
→ 系统卡死

# 操作系统：进程间消息传递
class MessageQueue:
    def send(self, pid, message):
        self.queue[pid].append(message)
    
    def receive(self, pid):
        return self.queue[pid].pop(0)

# Agent 世界：Agent 间消息传递
class AgentMessageBus:
    def send(self, from_agent, to_agent, message):
        self.queue[to_agent].append({
            "from": from_agent,
            "content": message,
            "timestamp": time.time()
        })
    
    def receive(self, agent_id):
        return self.queue[agent_id].pop(0)

Actor A  ──消息──→  Actor B 的 Mailbox
     ↑                    │
     └──── 响应消息 ───────┘

import asyncio

class SharedContext:
    def __init__(self):
        self._lock = asyncio.Lock()
        self._data = {}
    
    async def read(self, key):
        async with self._lock:
            return self._data.get(key)
    
    async def write(self, key, value):
        async with self._lock:
            self._data[key] = value

状态机：
WRITING → EDITING → APPROVED
            ↑
       (等待 Writer 完成)

用户程序（用户态）
     │
     │  write() 系统调用
     ↓
内核（内核态）
     │
     │  操作硬件
     ↓
磁盘 / 网卡 / 显示器

Agent（LLM 推理）
     │
     │  Function Call: search_web("...")
     ↓
Harness（执行环境）
     │
     │  执行工具
     ↓
搜索引擎 / 代码执行器 / 数据库

class Harness:
    def __init__(self):
        self.tools = {}      # 注册的工具
        self.policies = {}   # 安全策略
        self.resources = {}  # 资源配额
    
    async def execute(self, tool_call: ToolCall) -> ToolResult:
        # 1. 权限检查
        if not self._check_permission(tool_call):
            raise PermissionError("Not authorized")
        
        # 2. 资源配额检查
        if not self._check_quota(tool_call):
            raise ResourceExhaustedError("Quota exceeded")
        
        # 3. 执行工具
        tool = self.tools[tool_call.name]
        result = await tool.execute(tool_call.args)
        
        # 4. 审计日志
        self._log(tool_call, result)
        
        return result

// 操作系统内核处理系统调用
asmlinkage long sys_write(unsigned int fd, const char __user *buf, size_t count) {
    // 1. 权限检查（capable(CAP_SYS_WRITE)?）
    if (!capable(CAP_SYS_WRITE))
        return -EPERM;
    
    // 2. 参数验证（buf 是否用户空间指针？）
    if (!access_ok(buf, count))
        return -EFAULT;
    
    // 3. 执行业务逻辑
    return vfs_write(fd, buf, count, &pos);
    
    // 4. 审计
    audit_syscall_entry(...);
}

{
  "name": "search_web",
  "description": "搜索互联网获取信息",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "搜索关键词"
      },
      "max_results": {
        "type": "integer",
        "description": "最多返回结果数",
        "default": 10
      }
    },
    "required": ["query"]
  }
}

金字塔顶部（最快、最贵、最小）
━━━━━━━━━━━━━━━━━━━━━━
   寄存器    (~1 KB)    纳秒级
━━━━━━━━━━━━━━━━━━━━━━
   L1 Cache  (~32 KB)   纳秒级
━━━━━━━━━━━━━━━━━━━━━━
   L2 Cache  (~256 KB)  纳秒级
━━━━━━━━━━━━━━━━━━━━━━
   L3 Cache  (~8 MB)    十纳秒级
━━━━━━━━━━━━━━━━━━━━━━
   内存 RAM  (~16 GB)   百纳秒级
━━━━━━━━━━━━━━━━━━━━━━
   SSD       (~1 TB)    微秒级
━━━━━━━━━━━━━━━━━━━━━━
   磁盘      (~10 TB)   毫秒级
━━━━━━━━━━━━━━━━━━━━━━
金字塔底部（最慢、最便宜、最大）

# 操作系统内存管理
class MemoryManager:
    def __init__(self, physical_memory, disk):
        self.physical = physical_memory  # 快速但有限
        self.disk = disk                 # 慢速但海量
    
    def access(self, addr):
        if addr in TLB:  # Translation Lookaside Buffer
            return self.physical[addr]  # 命中，快！
        
        page = self.page_table[addr]
        if not page.in_memory:
            # 页面置换：把不常用的页换出到磁盘
            victim = self.find_victim_page()
            self.swap_out(victim)
            self.swap_in(page)
        
        return self.physical[addr]

class ContextManager:
    def __init__(self, max_tokens=128000):
        self.max_tokens = max_tokens
        self.messages = []
        self.summaries = {}  # 压缩后的摘要
    
    def add_message(self, role, content):
        self.messages.append({"role": role, "content": content})
        
        # 检查是否超出限制
        current_tokens = self.count_tokens(self.messages)
        
        if current_tokens > self.max_tokens * 0.8:
            # 触发压缩
            self._compress()
    
    def _compress(self):
        # 方法1：简单截断（丢弃最早的对话）
        while self.count_tokens(self.messages) > self.max_tokens * 0.5:
            self.messages.pop(0)
        
        # 方法2：智能摘要（保留关键信息，压缩冗余）
        # 用 LLM 本身生成摘要
        old_content = self._get_old_messages()
        summary = self.llm.summarize(f"请总结以下对话的要点：\n{old_content}")
        self.summaries["phase_1"] = summary
        
        # 方法3：分层记忆（工作记忆 vs 长期记忆）
        working = self.messages[-10:]  # 最近10轮
        longterm = self.summaries  # 历史摘要

┌─────────────────────────────────────┐
│  工作记忆（Working Memory）          │  ~4K tokens
│  当前任务、最近对话、临时变量        │
├─────────────────────────────────────┤
│  短期记忆（Short-term Memory）       │  ~32K tokens
│  今天/本周的会话、活跃项目上下文      │
├─────────────────────────────────────┤
│  长期记忆（Long-term Memory）        │  外部存储
│  压缩摘要、用户偏好、历史经验        │
└─────────────────────────────────────┘

/                    （根目录）
├── home/
│   └── user/
│       ├── documents/
│       │   ├── report.pdf
│       │   └── notes.txt
│       └── pictures/
│           └── photo.jpg
├── etc/
│   └── config.conf
└── tmp/
    └── cache.dat

# 把 U 盘挂载到 /mnt/usb
mount /dev/sdb1 /mnt/usb

# 把网络存储挂载到 /home/user/cloud
mount -t nfs 192.168.1.100:/share /home/user/cloud

# 把 ISO 镜像挂载到 /mnt/iso
mount -o loop image.iso /mnt/iso

┌─────────────────────────────────────────────┐
│  Agent 的"文件系统树"                        │
├─────────────────────────────────────────────┤
│  /context                    ← 当前对话      │
│  /memory                     ← 压缩记忆      │
│  /knowledge                  ← RAG 挂载点    │
│  │   ├── /company-docs       ← 公司文档     │
│  │   ├── /product-manual     ← 产品手册     │
│  │   └── /web-search-cache   ← 搜索缓存     │
│  /tools                     ← 可用工具       │
└─────────────────────────────────────────────┘

class RAGSystem:
    def __init__(self, vector_store, embedding_model):
        self.store = vector_store  # 向量数据库
        self.embed = embedding_model
    
    def retrieve(self, query: str, top_k: int = 5) -> List[Document]:
        # 1. 把查询向量化
        query_embedding = self.embed.encode(query)
        
        # 2. 在向量数据库中搜索相似文档
        results = self.store.similarity_search(
            query_embedding,
            top_k=top_k
        )
        
        return results
    
    def mount_to_context(self, query: str) -> str:
        # 1. 检索相关文档
        docs = self.retrieve(query)
        
        # 2. 组装成上下文
        context = "\n\n".join([doc.content for doc in docs])
        
        # 3. 添加到 prompt
        return f"参考信息：\n{context}\n\n用户问题：{query}"

┌─────────────────────────────────────────┐
│  热数据（Hot Storage）                    │  实时检索
│  常用文档、高频访问内容                   │
├─────────────────────────────────────────┤
│  温数据（Warm Storage）                  │  定期同步
│  近期文档、项目资料                       │
├─────────────────────────────────────────┤
│  冷数据（Cold Storage）                  │  按需加载
│  历史归档、罕见参考资料                   │
└─────────────────────────────────────────┘

# 简化的调度器逻辑
class Scheduler:
    def __init__(self):
        self.ready_queue = []      # 就绪队列
        self.running = None        # 当前运行进程
    
    def schedule(self):
        # 1. 把已完成的进程移出
        if self.running and self.running.is_terminated():
            self.running = None
        
        # 2. 把等待 I/O 的进程放回就绪队列
        for proc in self.waiting:
            if proc.is_io_complete():
                self.ready_queue.append(proc)
        
        # 3. 选择下一个进程运行
        if not self.running and self.ready_queue:
            self.running = self.ready_queue.pop(0)
            self.running.state = RUNNING
        
        # 4. 时间片用完，放回队列
        if self.running and self.running.used_full_timeslice():
            self.ready_queue.append(self.running)
            self.running.state = READY
            self.running = None

class Harness:
    def __init__(self, llm, tools, memory):
        self.llm = llm           # LLM = "CPU"
        self.tools = tools        # 工具 = "系统调用"
        self.memory = memory      # 记忆 = "内存"
        self.state = {}           # Agent 状态
        self.quota = Quota()     # 资源配额
    
    async def execute(self, task: str) -> str:
        # 1. 准备上下文
        context = await self._prepare_context(task)
        
        # 2. LLM 推理（可能多次）
        while not self._is_complete():
            # LLM 生成下一步
            action = await self.llm.think(context)
            
            # 如果是系统调用，执行工具
            if action.is_tool_call():
                result = await self._execute_tool(action)
                context.add_result(result)
            else:
                # 直接输出
                return action.content
        
        return context.final_answer()
    
    async def _execute_tool(self, tool_call: ToolCall):
        # 权限检查
        if not self._check_permission(tool_call):
            raise PermissionDenied()
        
        # 资源检查
        if not self.quota.can_use(tool_call):
            raise QuotaExceeded()
        
        # 执行
        tool = self.tools[tool_call.name]
        return await tool.execute(tool_call.args)

class Orchestrator:
    def __init__(self, agents: List[Agent], task_graph: Dict):
        self.agents = agents
        self.task_graph = task_graph  # 任务依赖关系
        self.results = {}
    
    async def run(self, root_task: str) -> Any:
        # 1. 构建执行计划（类似拓扑排序）
        execution_order = self._topological_sort(root_task)
        
        # 2. 按顺序执行，支持并行
        executed = set()
        
        while executed != set(execution_order):
            # 找出所有可以执行的 Agent（依赖已满足）
            ready = [
                agent for agent in self.agents
                if agent.name in execution_order
                and agent.name not in executed
                and all(
                    dep in executed 
                    for dep in self._get_dependencies(agent)
                )
            ]
            
            # 并行执行所有就绪的 Agent
            if ready:
                results = await asyncio.gather(*[
                    agent.execute(self.results) for agent in ready
                ])
                
                for agent, result in zip(ready, results):
                    self.results[agent.name] = result
                    executed.add(agent.name)
            else:
                # 死锁检测
                raise DeadlockError("No agents can proceed")
        
        return self.results[root_task]

// Capabilities 模型
struct process {
    cap_t cap;  // 进程的能力集
};

// 检查是否有写文件权限
if (!cap_raised(proc->cap, CAP_DAC_OVERRIDE)) {
    return -EACCES;
}

class ToolPermissions:
    def __init__(self):
        self.allowed_tools = set()
        self.rate_limits = {}
        self.data_scope = {}  # 数据访问范围
    
    def grant(self, tool: str, quota: dict):
        self.allowed_tools.add(tool)
        self.rate_limits[tool] = quota
    
    def check(self, agent: str, tool: str) -> bool:
        return tool in self.allowed_tools

# 使用示例
permissions = ToolPermissions()
permissions.grant("web_search", {"per_minute": 10, "per_day": 100})
permissions.grant("code_execution", {"per_hour": 50})
permissions.grant("read_files", {"/data/public/*": True, "/data/private/*": False})

// 信号处理
signal(SIGINT, handle_interrupt);  // 注册 Ctrl+C 处理函数

void handle_interrupt(int sig) {
    // 清理资源
    cleanup();
    exit(0);
}

class AgentEventSystem:
    def __init__(self):
        self.handlers = {}
        self.event_queue = asyncio.Queue()
    
    def on(self, event: str, handler: Callable):
        self.handlers[event] = handler
    
    async def emit(self, event: str, data: Any):
        await self.event_queue.put({"event": event, "data": data})
    
    async def process_events(self):
        while True:
            event = await self.event_queue.get()
            handler = self.handlers.get(event["event"])
            if handler:
                await handler(event["data"])

# 使用示例
events = AgentEventSystem()
events.on("tool_result", on_tool_result)
events.on("context_full", on_context_full)
events.on("agent_error", on_agent_error)

cat /proc/1234/maps      # 查看进程内存映射
cat /proc/1234/status    # 查看进程状态
cat /proc/cpuinfo        # CPU 信息
cat /proc/meminfo         # 内存信息

class AgentObserver:
    def __init__(self):
        self.metrics = {}
    
    def log(self, event: AgentEvent):
        self.metrics[event.type] = self.metrics.get(event.type, 0) + 1
    
    def get_state(self, agent_id: str) -> dict:
        return {
            "agent_id": agent_id,
            "current_task": self.get_current_task(agent_id),
            "context_tokens": self.count_tokens(agent_id),
            "tools_used": self.get_tools_used(agent_id),
            "execution_time": self.get_elapsed_time(agent_id),
            "errors": self.get_errors(agent_id)
        }
    
    def dump_trace(self, agent_id: str) -> str:
        # 返回完整的执行轨迹，用于调试
        return json.dumps(self.trace[agent_id], indent=2)

# Docker 的隔离
docker run --pid=host nginx        # 共享 PID 命名空间
docker run --network=host nginx     # 共享网络命名空间
docker run -v /host/path:/container/path nginx  # 共享挂载

class AgentSandbox:
    def __init__(self, isolation_level: str):
        self.isolation = isolation_level
    
    # 完全隔离：独立记忆和工具
    FULL = "full"
    
    # 共享工具，隔离记忆
    SHARED_TOOLS = "shared_tools"
    
    # 共享记忆，隔离工具
    SHARED_MEMORY = "shared_memory"
    
    # 完全共享（危险！）
    NONE = "none"