大语言模型应用开发：暂存工作空间设计与多智能体性能优化实践

在实际的大语言模型（LLM）应用开发中，我们常常面临一个看似简单却影响深远的工程问题：如何为模型提供一个干净、可控、可追溯的“工作空间”。无论是进行复杂的推理链（Chain-of-Thought）计算，还是执行需要多步工具调用的任务，模型都需要一个地方来存放中间状态、临时变量和计算过程。直接将所有信息都塞进上下文窗口，不仅会迅速耗尽宝贵的Token，还会让系统状态变得混乱不堪，难以调试和优化。这就是“Scratch Workspaces”（暂存工作空间）概念的核心价值。

对于开发者而言，为LLM设计一个高效的暂存工作空间，意味着能将模型的“思考过程”与最终的“输出结果”分离。这不仅仅是节省Token，更是构建可解释、可迭代、高性能的LLM应用架构的关键。本文将深入探讨为什么LLM需要暂存工作空间，如何从零开始设计一个满足生产需求的工作空间，并通过一个具体的多智能体（Multi-Agent）协作场景，展示如何利用工作空间来优化服务延迟和性能。我们将重点关注如何将最新的研究理念，如面向异构LLM的延迟与性能感知的多智能体服务（latency- and performance-aware multi-agent serving for heterogeneous llms），落地为可实践的工程方案。

1. 理解暂存工作空间：为什么它比直接使用上下文更有效

在深入代码之前，我们必须先厘清一个根本问题：既然LLM的上下文窗口可以存储信息，为什么还需要一个独立的工作空间？

1.1 上下文窗口的局限性

LLM的上下文窗口（Context Window）本质是一个固定长度的序列，模型通过注意力机制处理其中的所有Token。虽然它可以存储对话历史、系统提示和用户输入，但将其用作“工作空间”存在几个致命缺陷：

• 成本高昂：每次模型推理，上下文中的所有Token都需要参与计算，直接推高了API调用成本（对于按Token计费的云服务）或计算资源消耗（对于本地部署）。
• 状态污染：中间推理步骤、失败的尝试、工具调用的详细日志如果全部保留在上下文中，会成为后续推理的“噪声”，可能干扰模型的判断。
• 难以追溯与调试：当任务出错时，你需要从冗长的上下文历史中筛选出关键的中间步骤，这非常低效。一个独立的工作空间可以结构化地记录每一步的输入、输出和状态。
• 长度限制：尽管上下文窗口在增大，但它终究是有限的。复杂任务（如代码生成、长文档分析）的中间过程很容易触及这个上限。

1.2 暂存工作空间的核心优势

暂存工作空间是一个与主上下文分离的、由应用程序管理的存储区域。它的设计遵循几个核心原则：

1. 状态外置：将模型的“思考草稿”移出昂贵的上下文，放入应用层的内存或数据库中。
2. 结构化存储：工作空间内的数据不是纯文本，而是结构化的对象（如JSON），包含步骤ID、类型、输入、输出、时间戳、状态（成功/失败）等元数据。
3. 按需加载：只有在当前步骤确实需要参考之前某一步的结果时，才将那个特定结果的精简摘要加载到上下文中，而不是加载整个历史。
4. 生命周期管理：工作空间可以针对单个会话（Session）或单个任务（Task）创建，任务完成后即可清理，资源利用率高。

这种模式使得LLM应用能够处理更复杂、更长的任务链，同时保持可观察性（Observability）和可控性。

2. 设计一个生产可用的暂存工作空间

理解了“为什么”之后，我们开始设计“怎么做”。一个最小可用的工作空间需要包含数据模型、存储层和操作接口。

2.1 定义工作空间的数据模型

我们首先用Python的Pydantic模型来定义工作空间中的基本单元：WorkSpaceStep（步骤）。这个模型将确保我们数据的结构化和类型安全。

from pydantic import BaseModel, Field from enum import Enum from datetime import datetime from typing import Any, Optional, Dict class StepStatus(str, Enum): """步骤执行状态枚举""" PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" CANCELLED = "cancelled" class StepType(str, Enum): """步骤类型枚举，便于路由和处理""" LLM_REASONING = "llm_reasoning" # LLM推理步骤 TOOL_CALL = "tool_call" # 工具调用步骤 CODE_EXECUTION = "code_execution" # 代码执行步骤 CONDITIONAL_BRANCH = "conditional_branch" # 条件分支判断 class WorkSpaceStep(BaseModel): """工作空间中的单个步骤记录""" step_id: str = Field(..., description="步骤唯一标识符") task_id: str = Field(..., description="所属任务ID") parent_step_id: Optional[str] = Field(None, description="父步骤ID，用于构建步骤树") step_type: StepType = Field(..., description="步骤类型") # 输入与输出 input_data: Dict[str, Any] = Field(default_factory=dict, description="步骤输入数据") output_data: Optional[Dict[str, Any]] = Field(None, description="步骤输出数据") error_info: Optional[str] = Field(None, description="如果失败，错误信息") # 状态与元数据 status: StepStatus = Field(default=StepStatus.PENDING, description="步骤状态") created_at: datetime = Field(default_factory=datetime.now) started_at: Optional[datetime] = None completed_at: Optional[datetime] = None # 性能与成本追踪（对异构LLM服务至关重要） llm_model_used: Optional[str] = Field(None, description="使用的LLM模型名称") prompt_tokens: Optional[int] = Field(None, description="提示词消耗的Token数") completion_tokens: Optional[int] = Field(None, description="回复消耗的Token数") latency_ms: Optional[float] = Field(None, description="步骤执行延迟（毫秒）") class Config: use_enum_values = True # 序列化时使用枚举的值

这个模型定义了每个步骤的完整生命周期。llm_model_used和latency_ms等字段是为后续实现性能感知的多智能体调度所做的预留。

2.2 选择存储后端并实现管理层

对于学习环境或中等负载，我们可以使用内存存储（如字典）或轻量级数据库（如SQLite）。生产环境则需要考虑Redis（用于高速缓存）、PostgreSQL或MongoDB（用于持久化）。

下面是一个基于内存字典的简单工作空间管理器，它演示了核心接口：

from typing import List, Optional import uuid import asyncio from contextlib import asynccontextmanager class ScratchWorkSpaceManager: """暂存工作空间管理器（内存版本）""" def __init__(self): # 内存存储：task_id -> List[WorkSpaceStep] self._task_steps: Dict[str, List[WorkSpaceStep]] = {} # 步骤索引：step_id -> WorkSpaceStep self._step_index: Dict[str, WorkSpaceStep] = {} def create_task(self, task_id: Optional[str] = None) -> str: """创建一个新的任务工作空间""" if task_id is None: task_id = f"task_{uuid.uuid4().hex[:8]}" if task_id not in self._task_steps: self._task_steps[task_id] = [] return task_id def add_step(self, task_id: str, step: WorkSpaceStep) -> None: """向指定任务添加一个步骤""" if task_id not in self._task_steps: raise ValueError(f"Task {task_id} not found") self._task_steps[task_id].append(step) self._step_index[step.step_id] = step def get_step(self, step_id: str) -> Optional[WorkSpaceStep]: """根据ID获取步骤""" return self._step_index.get(step_id) def get_task_steps(self, task_id: str) -> List[WorkSpaceStep]: """获取一个任务的所有步骤（按创建时间排序）""" steps = self._task_steps.get(task_id, []) return sorted(steps, key=lambda x: x.created_at) def update_step_status(self, step_id: str, status: StepStatus, output: Optional[Dict] = None, error: Optional[str] = None, **metrics) -> bool: """更新步骤状态、输出和性能指标""" step = self._step_index.get(step_id) if not step: return False step.status = status if status == StepStatus.RUNNING and not step.started_at: step.started_at = datetime.now() elif status in [StepStatus.SUCCESS, StepStatus.FAILED, StepStatus.CANCELLED]: step.completed_at = datetime.now() if output is not None: step.output_data = output if error is not None: step.error_info = error # 更新性能指标，如延迟、Token数等 for key, value in metrics.items(): if hasattr(step, key): setattr(step, key, value) return True @asynccontextmanager async def step_context(self, task_id: str, step_type: StepType, input_data: Dict, parent_step_id: Optional[str] = None): """一个异步上下文管理器，用于自动管理步骤的生命周期和计时""" step_id = f"step_{uuid.uuid4().hex[:8]}" step = WorkSpaceStep( step_id=step_id, task_id=task_id, parent_step_id=parent_step_id, step_type=step_type, input_data=input_data, status=StepStatus.PENDING ) self.add_step(task_id, step) # 标记为运行中 self.update_step_status(step_id, StepStatus.RUNNING) start_time = asyncio.get_event_loop().time() try: yield step_id # 将step_id交给调用者使用 # 如果上下文块正常退出，标记为成功（输出需由调用者稍后更新） # 注意：成功状态和输出通常在具体操作完成后明确更新 except Exception as e: # 如果发生异常，标记为失败 self.update_step_status(step_id, StepStatus.FAILED, error=str(e)) raise finally: # 计算并记录延迟 end_time = asyncio.get_event_loop().time() latency_ms = (end_time - start_time) * 1000 # 如果状态还未被更新为成功或失败（比如在yield后发生了异常但被捕获），这里保持原状态 current_step = self.get_step(step_id) if current_step and current_step.status == StepStatus.RUNNING: # 如果还在运行状态，可能意味着yield后的代码没更新状态，这是一个安全兜底 pass # 更新延迟指标，其他指标（如tokens）在具体LLM调用后更新 self.update_step_status(step_id, current_step.status, latency_ms=latency_ms)

这个管理器提供了工作空间的核心CRUD操作。step_context上下文管理器是一个关键设计，它确保了每个步骤的开始、结束、异常捕获和性能指标（如延迟）的自动记录，这大大简化了业务代码。

3. 在异构多智能体服务场景中应用工作空间

现在，我们将工作空间与“延迟与性能感知的多智能体服务”这一前沿理念结合。假设我们有一个任务：分析一份公司财报，并生成一份投资建议摘要。这个任务可以分解为：

1. 智能体A（大模型，能力强，速度慢）：提取关键财务数据。
2. 智能体B（小模型，能力专，速度快）：进行同比/环比计算。
3. 智能体C（大模型）：根据计算结果生成叙事性摘要。

我们的目标是：利用工作空间协调这三个智能体，并基于性能（延迟、成本）做出动态决策。

3.1 定义智能体与路由逻辑

首先，我们抽象一个智能体基类，并实现两个具体的LLM智能体（模拟不同性能）。

import aiohttp import json import random from abc import ABC, abstractmethod class LLMAgent(ABC): """LLM智能体抽象基类""" def __init__(self, name: str, model: str, base_url: str, api_key: str = ""): self.name = name self.model = model self.base_url = base_url self.api_key = api_key # 模拟性能特征：平均延迟(ms)和每千Token成本(美分) self.performance_profile = {"avg_latency": 1000, "cost_per_1k_tokens": 10.0} @abstractmethod async def invoke(self, prompt: str, system_prompt: str = "") -> Dict[str, Any]: """调用智能体，返回包含输出和元数据的字典""" pass class MockPowerfulAgent(LLMAgent): """模拟一个能力强但延迟高的大模型智能体（如GPT-4级别）""" def __init__(self): super().__init__(name="Agent-Powerful", model="gpt-4", base_url="https://api.openai.com/v1") self.performance_profile = {"avg_latency": 2000, "cost_per_1k_tokens": 30.0} async def invoke(self, prompt: str, system_prompt: str = "") -> Dict[str, Any]: # 模拟网络延迟和思考时间 await asyncio.sleep(random.uniform(1.5, 2.5)) # 模拟一个复杂的、高质量的回复 mock_response = f"[基于强大模型的分析] {prompt[:50]}... 的关键指标是：营收增长15%，利润率提升2%。" input_tokens = len(prompt) // 4 output_tokens = len(mock_response) // 4 return { "content": mock_response, "input_tokens": input_tokens, "output_tokens": output_tokens, "model": self.model } class MockFastAgent(LLMAgent): """模拟一个速度快、成本低但能力侧重的小模型智能体（如Claude Haiku）""" def __init__(self): super().__init__(name="Agent-Fast", model="claude-3-haiku", base_url="https://api.anthropic.com/v1") self.performance_profile = {"avg_latency": 500, "cost_per_1k_tokens": 0.25} async def invoke(self, prompt: str, system_prompt: str = "") -> Dict[str, Any]: # 模拟更快的响应 await asyncio.sleep(random.uniform(0.3, 0.8)) # 模拟一个快速、直接的回复，可能深度不足 if "计算" in prompt: mock_response = "计算完成：同比增长率为15%。" else: mock_response = f"[快速回复] 已处理：{prompt[:30]}..." input_tokens = len(prompt) // 4 output_tokens = len(mock_response) // 4 return { "content": mock_response, "input_tokens": input_tokens, "output_tokens": output_tokens, "model": self.model }

接下来，实现一个简单的延迟与性能感知的路由器。这个路由器根据任务类型、当前系统负载（或预算）以及智能体的历史性能数据，决定将任务分发给哪个智能体。

class PerformanceAwareRouter: """性能感知路由器""" def __init__(self): self.agents = { 'powerful': MockPowerfulAgent(), 'fast': MockFastAgent() } # 可以在此处集成更复杂的负载均衡器或成本控制器 def select_agent(self, step_type: StepType, priority: str = "balanced") -> LLMAgent: """ 根据步骤类型和优先级选择智能体。 priority: 'speed', 'quality', 'balanced', 'cost' """ if step_type == StepType.LLM_REASONING: if priority == "speed": return self.agents['fast'] elif priority == "quality": return self.agents['powerful'] else: # balanced or cost # 简单策略：需要深度推理用大模型，简单任务用小模型 # 实际项目中，这里可以接入一个预测模型或规则引擎 return self.agents['powerful'] # 示例默认 elif step_type == StepType.TOOL_CALL: # 工具调用可能更注重速度和确定性 return self.agents['fast'] else: # 默认返回快速智能体 return self.agents['fast']

3.2 构建基于工作空间的任务执行引擎

这是最核心的部分：一个协调器（Orchestrator），它读取工作空间中的任务步骤，根据步骤类型调用相应的智能体或工具，并将结果写回工作空间。

class TaskOrchestrator: """任务协调器，连接工作空间、路由器和具体执行单元""" def __init__(self, workspace: ScratchWorkSpaceManager, router: PerformanceAwareRouter): self.workspace = workspace self.router = router async def execute_task(self, task_id: str, initial_input: str): """执行一个多步骤任务""" print(f"[Orchestrator] 开始执行任务: {task_id}") # 步骤1: 使用强大模型进行深度分析 step1_input = {"prompt": f"请分析以下财报文本，提取关键财务数据（营收、利润、增长率）:\n{initial_input}"} async with self.workspace.step_context(task_id, StepType.LLM_REASONING, step1_input) as step1_id: agent = self.router.select_agent(StepType.LLM_REASONING, priority="quality") print(f"[Step1] 选择智能体: {agent.name} 进行深度分析") result = await agent.invoke(step1_input["prompt"], "你是一个财务分析师。") # 更新步骤结果和性能指标 self.workspace.update_step_status( step1_id, StepStatus.SUCCESS, output={"analysis": result["content"]}, llm_model_used=agent.model, prompt_tokens=result["input_tokens"], completion_tokens=result["output_tokens"] ) analysis_result = result["content"] # 步骤2: 使用快速模型进行计算 step2_input = {"prompt": f"基于以下分析‘{analysis_result[:100]}’，计算主要的同比增长率。"} async with self.workspace.step_context(task_id, StepType.LLM_REASONING, step2_input, parent_step_id=step1_id) as step2_id: agent = self.router.select_agent(StepType.LLM_REASONING, priority="speed") print(f"[Step2] 选择智能体: {agent.name} 进行快速计算") result = await agent.invoke(step2_input["prompt"]) self.workspace.update_step_status( step2_id, StepStatus.SUCCESS, output={"calculation": result["content"]}, llm_model_used=agent.model, prompt_tokens=result["input_tokens"], completion_tokens=result["output_tokens"] ) calc_result = result["content"] # 步骤3: 综合前两步结果，生成最终摘要 step3_input = {"prompt": f"财务分析摘要：{analysis_result}\n关键计算：{calc_result}\n请生成一份给投资者的简要建议报告。"} async with self.workspace.step_context(task_id, StepType.LLM_REASONING, step3_input, parent_step_id=step2_id) as step3_id: agent = self.router.select_agent(StepType.LLM_REASONING, priority="balanced") print(f"[Step3] 选择智能体: {agent.name} 生成最终报告") result = await agent.invoke(step3_input["prompt"], "你是一名投资顾问。") self.workspace.update_step_status( step3_id, StepStatus.SUCCESS, output={"final_report": result["content"]}, llm_model_used=agent.model, prompt_tokens=result["input_tokens"], completion_tokens=result["output_tokens"] ) final_output = result["content"] print(f"[Orchestrator] 任务 {task_id} 执行完毕。") print(f"最终报告:\n{final_output}") # 返回最终结果和所有步骤记录，用于展示或持久化 all_steps = self.workspace.get_task_steps(task_id) return final_output, all_steps

3.3 运行与验证

现在，我们可以将上述所有组件组合起来，运行一个完整的示例。

import asyncio async def main(): # 1. 初始化组件 workspace = ScratchWorkSpaceManager() router = PerformanceAwareRouter() orchestrator = TaskOrchestrator(workspace, router) # 2. 创建任务 task_id = workspace.create_task() # 3. 模拟输入（财报文本） financial_report_text = """ 某某科技公司2023年第四季度财报显示： 总营收为120亿元，去年同期为100亿元。 净利润为25亿元，去年同期为20亿元。 研发投入为18亿元，销售与市场费用为15亿元。 """ # 4. 执行任务 final_report, all_steps = await orchestrator.execute_task(task_id, financial_report_text) # 5. 展示工作空间记录（用于调试和监控） print("\n" + "="*50) print("工作空间步骤详情:") print("="*50) for step in all_steps: print(f"[{step.step_id}] {step.step_type.value} - {step.status.value}") print(f" 模型: {step.llm_model_used or 'N/A'}, 延迟: {step.latency_ms:.1f}ms") print(f" 输入: {str(step.input_data)[:80]}...") if step.output_data: print(f" 输出: {str(step.output_data)[:80]}...") print() if __name__ == "__main__": asyncio.run(main())

运行这段代码，你将看到任务被分解为三个步骤执行，每个步骤根据其类型和我们的简单路由策略，动态选择了不同的智能体（模拟）。所有步骤的输入、输出、状态、使用的模型和延迟都被清晰地记录在工作空间中。

4. 生产环境进阶：从原型到可部署系统

上面的示例是一个概念验证。要将暂存工作空间和性能感知多智能体服务用于生产，还需要考虑以下关键点。

4.1 存储后端的升级与选型

内存存储无法持久化，也无法在多实例间共享。生产环境需要更健壮的方案。

实现建议：定义一个抽象的WorkSpaceStorage接口，然后为不同的后端提供实现。这样可以在开发和生产环境之间轻松切换。

from abc import ABC, abstractmethod class WorkSpaceStorage(ABC): @abstractmethod async def save_step(self, step: WorkSpaceStep) -> bool: pass @abstractmethod async def get_step(self, task_id: str, step_id: str) -> Optional[WorkSpaceStep]: pass @abstractmethod async def get_task_steps(self, task_id: str) -> List[WorkSpaceStep]: pass # ... 其他方法

4.2 性能感知路由的复杂策略

示例中的路由器非常简单。真实场景中，路由策略需要考虑更多维度：

1. 动态性能监控：实时收集每个智能体（或模型端点）的延迟、成功率、错误率，而不仅仅是静态配置。
2. 成本控制：为每个任务或用户设置Token预算，在预算内选择最合适的模型。
3. 负载均衡：避免将所有请求打到同一个模型实例上。
4. Fallback机制：当首选智能体失败或超时时，自动切换到备用智能体。
5. 基于内容的路由：根据输入文本的长度、语言、领域（如代码、法律）选择专精模型。

一个进阶的路由器可能维护一个AgentHealth看板，并实现如下决策逻辑：

class AdvancedRouter: async def select_agent(self, step_context: StepContext) -> LLMAgent: candidates = self._get_available_agents(step_context.required_capability) # 策略1: 成本优先（在预算内选最快的） if step_context.budget_priority == "cost": candidates = sorted(candidates, key=lambda a: (a.current_cost_per_1k_tokens, a.avg_latency)) # 策略2: 延迟敏感（选当前延迟最低的） elif step_context.budget_priority == "latency": candidates = sorted(candidates, key=lambda a: (a.current_latency_estimate, a.current_cost_per_1k_tokens)) # 策略3: 质量优先（选能力最强的，忽略成本） else: # "quality" candidates = sorted(candidates, key=lambda a: (-a.capability_score, a.avg_latency)) # 加入随机扰动，避免雪崩 selected = self._add_jitter(candidates[:3]) return selected

4.3 工作空间的观测性与调试

工作空间存储了完整的执行轨迹，这是强大的调试工具。你需要提供方便的方式来查询和可视化这些数据。

• API端点：提供GET /tasks/{task_id}/steps和GET /steps/{step_id}等API，方便前端或监控系统查看。
• 日志集成：将关键步骤（开始、成功、失败）与应用的日志系统（如ELK、Loki）关联，并赋予唯一的追踪ID（trace_id）。
• 可视化界面：开发一个简单的内部面板，以时间线或树形图展示任务步骤，高亮显示失败步骤和性能瓶颈。

4.4 错误处理与重试机制

在生产中，LLM调用和工具调用都可能失败。工作空间需要支持复杂的错误处理流程。

• 步骤重试：对于因网络抖动导致的失败，可以自动重试。重试次数和退避策略应可配置。
• 条件分支：根据步骤的成功/失败输出，动态决定下一步执行哪个分支。这需要扩展StepType，支持CONDITIONAL_BRANCH类型。
• 补偿操作：如果一个步骤失败且无法重试，可能需要执行一些清理或状态回滚操作。这些逻辑也可以作为特殊步骤记录在工作空间中。

5. 常见问题与排查路径

在实现和使用暂存工作空间时，你可能会遇到以下典型问题。

6. 最佳实践与扩展方向

6.1 实施清单

在将暂存工作空间方案引入项目前，请对照此清单：

• [ ]明确边界：定义清楚哪些数据放上下文，哪些放工作空间。一般规则：最终输出和极简的当前指令放上下文；中间过程、历史记录、元数据放工作空间。
• [ ]设计数据结构：在项目早期用Pydantic或类似工具定义好WorkSpaceStep的数据模型，避免后期字段混乱。
• [ ]选择存储：根据数据量、查询模式和团队技术栈，选择Redis、PostgreSQL或MongoDB。从简单开始，但预留切换接口。
• [ ]集成观测：在添加步骤、更新状态、记录指标的关键位置，打入日志和Metrics（如Prometheus），便于监控。
• [ ]设计清理策略：确定工作空间数据的保留策略（例如，成功任务保留7天，失败任务保留30天），并实现自动化清理任务。
• [ ]编写单元测试：为工作空间管理器的核心方法（创建、添加、更新、查询）编写单元测试，确保数据一致性。

6.2 扩展方向

1. 版本化工作空间：支持步骤的版本回滚。当某一步的LLM生成结果不理想时，可以回退到上一步，用不同的参数重新执行，而无需重跑整个任务。
2. 工作空间快照与共享：允许将某个任务的工作空间状态保存为“模板”或“快照”，并分享给其他任务或用户复用，这对于调试和知识沉淀非常有价值。
3. 与向量数据库结合：将工作空间中成功的步骤输出（如高质量的分析、代码片段）存入向量数据库，构建一个“内部最佳实践库”。新的任务可以首先检索相似的成功案例作为参考。
4. 实现真正的分布式协调：当智能体本身是分布式的微服务时，工作空间需要成为一个共享的、一致的状态存储。可以考虑使用分布式锁（如Redis Redlock）和事务来保证多节点写入的一致性。
5. 自动化评估与优化：利用工作空间中记录的输入、输出和性能指标，训练一个评估模型（Evaluator），自动判断任务执行的质量和效率，并反馈给路由器，形成闭环优化。

暂存工作空间不是一个炫技的概念，而是一个解决LLM应用工程中实际痛点的架构模式。它通过将“思考过程”外置和结构化，为复杂任务编排、性能优化、成本控制和系统观测提供了坚实的基础。从简单的内存字典开始，逐步演进到支持多智能体、性能感知、持久化存储的生产级系统，这条路径清晰且回报显著。当你下一次设计一个需要多步推理或工具调用的LLM应用时，不妨先问自己：我的模型，有一个足够好的“草稿纸”吗？