BAML框架实战：LLM调用层标准化迁移与智能体系统优化

在构建基于LLM的智能体系统时，很多开发者都遇到过这样的困境：随着业务逻辑复杂度的提升，LLM调用层的代码变得臃肿且难以维护，不同模型提供商的API差异、复杂的参数配置、错误处理逻辑散落在各个角落。最近我在Agentique项目中就面临了这个问题，经过技术选型后，决定将整个LLM层迁移到BAML框架。本文将完整分享这次迁移的技术细节、实战步骤和避坑经验。

无论你是正在构建自己的LLM应用，还是对现有的AI架构进行优化，这篇文章都将为你提供一套完整的解决方案。我们将从BAML框架的核心概念入手，逐步演示如何将复杂的LLM调用逻辑标准化、模块化，最终实现更高效、更稳定的智能体系统。

1. LLM调用层的挑战与BAML解决方案

1.1 传统LLM调用层的问题分析

在典型的LLM应用开发中，开发者通常需要直接调用各种模型提供商的SDK或API。以OpenAI、Anthropic、本地部署模型为例，每个提供商都有自己独特的参数格式、错误码体系和速率限制策略。

# 传统方式 - 散落的LLM调用代码 import openai from anthropic import Anthropic def call_openai(prompt, model="gpt-4"): try: response = openai.ChatCompletion.create( model=model, messages=[{"role": "user", "content": prompt}], temperature=0.7, max_tokens=1000 ) return response.choices[0].message.content except openai.error.RateLimitError: # 重试逻辑 pass except openai.error.APIError: # 错误处理 pass def call_anthropic(prompt, model="claude-3-sonnet"): client = Anthropic(api_key=ANTHROPIC_API_KEY) try: message = client.messages.create( model=model, max_tokens=1000, messages=[{"role": "user", "content": prompt}] ) return message.content except Exception as e: # 不同的错误处理 pass

这种方式的痛点显而易见：代码重复、错误处理不一致、配置分散、难以扩展新模型。当业务需要支持多个模型提供商或进行A/B测试时，维护成本呈指数级增长。

1.2 BAML框架的核心价值

BAML（Builders' AI Model Language）是一个专门为LLM应用设计的声明式框架，它通过统一的DSL（领域特定语言）来描述LLM调用规范，实现了以下核心价值：

• 标准化接口：统一的模型调用抽象，屏蔽不同提供商的API差异
• 类型安全：强类型的输入输出定义，减少运行时错误
• 配置集中化：所有模型配置、参数调优在单一位置管理
• 可测试性：内置的测试框架和Mock支持
• 可维护性：清晰的架构分层，便于团队协作

1.3 Agentique项目背景

Agentique是一个基于LLM的多智能体协作框架，原本的LLM调用层采用了传统的直接调用方式。随着智能体类型的增加和业务场景的复杂化，出现了以下典型问题：

1. 新增模型支持需要修改多处代码
2. 错误处理逻辑不一致导致系统稳定性下降
3. 配置管理混乱，不同环境下的模型行为差异大
4. 缺乏统一的性能监控和日志记录

迁移到BAML正是为了解决这些工程化难题，提升整个系统的可维护性和扩展性。

2. 环境准备与BAML基础配置

2.1 系统环境要求

在进行迁移之前，需要确保开发环境满足以下要求：

• Python 3.8+ 或 Node.js 16+
• 包管理工具：pip 或 npm
• 版本控制：Git
• 推荐的IDE：VS Code with BAML扩展

2.2 BAML安装与初始化

首先安装BAML核心包和相关的开发工具：

# 对于Python项目 pip install baml-core baml-dev # 对于TypeScript项目 npm install @baml/core @baml/dev

初始化BAML配置文件，创建项目结构：

# 在项目根目录执行 baml init

这会生成以下目录结构：

project-root/ ├── baml_src/ │ ├── models/ │ │ └── chat.baml │ ├── functions/ │ │ └── agent_actions.baml │ └── types/ │ └── common.baml ├── baml.config.yaml ├── generated/ │ ├── python/ │ └── typescript/ └── tests/ └── baml_tests/

2.3 基础配置详解

baml.config.yaml是BAML的核心配置文件，定义了模型提供商、认证信息和全局参数：

version: 1 project: name: "agentique-llm-layer" language: ["python", "typescript"] models: - name: "openai-gpt4" provider: "openai" config: model: "gpt-4" api_key: "${OPENAI_API_KEY}" temperature: 0.7 max_tokens: 1000 - name: "anthropic-claude" provider: "anthropic" config: model: "claude-3-sonnet-20240229" api_key: "${ANTHROPIC_API_KEY}" max_tokens: 1000 - name: "local-llama" provider: "ollama" config: model: "llama2:13b" base_url: "http://localhost:11434" runtime: retry: max_attempts: 3 backoff_multiplier: 2 timeout: 30000 circuit_breaker: failure_threshold: 5 reset_timeout: 60000 logging: level: "info" format: "json"

这个配置文件集中管理了所有LLM相关的设置，包括重试策略、超时控制和熔断机制，为后续的迁移工作奠定了基础。

3. BAML核心概念与语法详解

3.1 类型定义系统

BAML使用强类型系统来定义LLM交互的数据结构，这大大提高了代码的可靠性和可维护性。在baml_src/types/common.baml中定义基础类型：

// 定义智能体消息类型 type AgentMessage { role: "user" | "assistant" | "system" content: string timestamp: datetime metadata?: Map<string, string> } // 定义LLM响应类型 type LLMResponse { content: string model: string usage: { prompt_tokens: int completion_tokens: int total_tokens: int } latency: float } // 定义错误类型 type LLMError { code: string message: string retryable: boolean details?: Map<string, any> }

类型系统支持联合类型、可选字段、映射等高级特性，确保在编译时就能发现数据结构的错误。

3.2 函数定义与模型绑定

在baml_src/functions/agent_actions.baml中定义LLM调用函数：

// 定义基础的聊天完成函数 function ChatCompletion { input { messages: AgentMessage[] model: "openai-gpt4" | "anthropic-claude" | "local-llama" temperature?: float between 0.0 and 2.0 max_tokens?: int between 1 and 4000 } output { response: LLMResponse error?: LLMError } // 绑定到具体的模型配置 impl<LLM> using model { prompt: """ {{#each messages}} <{{role}}>{{content}}</{{role}}> {{/each}} """ config { temperature: temperature ?? 0.7 max_tokens: max_tokens ?? 1000 } } }

3.3 模板系统与提示词管理

BAML提供了强大的模板系统，支持条件逻辑、循环和局部模板，使得提示词管理变得简单而高效：

// 在 baml_src/templates/agent_system_prompt.baml 中 template AgentSystemPrompt for AgentType { when AgentType is "analyzer" { """ 你是一个专业的数据分析助手，擅长从复杂信息中提取关键洞察。 你的任务要求： 1. 准确理解用户的数据分析需求 2. 提供结构化的分析结果 3. 指出数据中的异常模式和趋势 4. 给出可行的行动建议 请确保回答专业、准确、有洞察力。 """ } when AgentType is "coder" { """ 你是一个经验丰富的软件开发助手，精通多种编程语言和框架。 你的编码原则： 1. 代码要简洁、可读、可维护 2. 遵循行业最佳实践和编码规范 3. 考虑性能和安全性 4. 提供清晰的注释和文档 请确保代码正确、高效、符合要求。 """ } }

4. Agentique LLM层迁移实战

4.1 迁移策略与步骤规划

将Agentique的LLM层迁移到BAML需要系统性的方法，我们采用分阶段迁移策略：

1. 分析阶段：梳理现有的LLM调用点和依赖关系
2. 设计阶段：设计BAML类型系统和函数接口
3. 实现阶段：逐步替换原有调用，保持向后兼容
4. 测试阶段：确保功能一致性和性能达标
5. 优化阶段：利用BAML特性进行架构优化

4.2 现有代码分析与人机交互模式识别

首先分析Agentique中现有的LLM调用模式：

# 原有的智能体基类中的LLM调用方法 class BaseAgent: def __init__(self, model_provider="openai", model_config=None): self.model_provider = model_provider self.model_config = model_config or {} self.setup_llm_client() def setup_llm_client(self): if self.model_provider == "openai": self.llm_client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY")) elif self.model_provider == "anthropic": self.llm_client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY")) # 其他提供商... def call_llm(self, messages, **kwargs): """统一的LLM调用方法，但内部实现复杂""" if self.model_provider == "openai": return self._call_openai(messages, **kwargs) elif self.model_provider == "anthropic": return self._call_anthropic(messages, **kwargs) # 其他提供商... def _call_openai(self, messages, **kwargs): # 复杂的参数处理和错误处理逻辑 try: response = self.llm_client.chat.completions.create( model=self.model_config.get("model", "gpt-4"), messages=messages, temperature=kwargs.get("temperature", 0.7), max_tokens=kwargs.get("max_tokens", 1000) ) return self._parse_openai_response(response) except Exception as e: return self._handle_error(e) # 其他提供商的实现方法...

通过分析发现，原有的代码中存在大量的重复逻辑和供应商特定的处理代码，这正是BAML要解决的核心问题。

4.3 BAML类型系统设计与实现

根据Agentique的业务需求，设计对应的BAML类型系统：

// baml_src/types/agentique.baml type AgentRequest { agent_id: string session_id: string messages: AgentMessage[] context: { environment: "development" | "staging" | "production" user_preferences?: Map<string, any> system_constraints?: string[] } } type AgentResponse { success: boolean content: string agent_actions?: AgentAction[] metadata: { model_used: string processing_time: float tokens_used: int cost_estimate: float } } type AgentAction { type: "tool_call" | "api_request" | "database_query" | "user_prompt" name: string parameters: Map<string, any> description: string }

4.4 核心函数迁移与适配器模式实现

创建BAML函数来替代原有的LLM调用逻辑：

// baml_src/functions/agentique_core.baml function ProcessAgentRequest { input { request: AgentRequest model_preference?: string } output { response: AgentResponse } impl<LLM> using model_preference ?? "openai-gpt4" { prompt: """ {{> agent_system_prompt agent_type=request.context.agent_type}} 对话历史： {{#each request.messages}} {{role}}: {{content}} {{/each}} 当前会话上下文： - 环境: {{request.context.environment}} - 用户偏好: {{request.context.user_preferences}} - 系统约束: {{request.context.system_constraints}} 请根据以上信息生成合适的响应，并决定是否需要执行额外的动作。 """ config { temperature: 0.7 max_tokens: 1500 } } }

同时创建适配器层，确保平滑迁移：

# src/llm/adapters/legacy_adapter.py from typing import Dict, Any, List import os from baml import BAMLClient from ..models.agent_models import AgentRequest, AgentResponse class LegacyToBAMLAdapter: """将旧版LLM调用适配到BAML接口""" def __init__(self): self.baml = BAMLClient() def convert_legacy_messages(self, legacy_messages: List[Dict]) -> List[Dict]: """将旧版消息格式转换为BAML标准格式""" converted = [] for msg in legacy_messages: converted.append({ "role": msg.get("role", "user"), "content": msg.get("content", ""), "timestamp": msg.get("timestamp"), "metadata": msg.get("metadata", {}) }) return converted async def process_agent_request(self, agent_id: str, messages: List[Dict], context: Dict[str, Any] = None) -> Dict[str, Any]: """处理智能体请求 - 新版BAML接口""" request = AgentRequest( agent_id=agent_id, session_id=context.get("session_id", "default"), messages=self.convert_legacy_messages(messages), context={ "environment": context.get("environment", "development"), "user_preferences": context.get("user_preferences", {}), "system_constraints": context.get("system_constraints", []) } ) # 使用BAML处理请求 response = await self.baml.ProcessAgentRequest( request=request, model_preference=context.get("model_preference") ) # 转换为旧版响应格式以保持兼容性 return { "success": response.success, "content": response.content, "actions": [action.dict() for action in response.agent_actions] if response.agent_actions else [], "metadata": response.metadata.dict() }

4.5 渐进式迁移与双模式运行

为了确保迁移过程平稳，我们实现双模式运行机制：

# src/llm/llm_manager.py from enum import Enum from typing import Dict, Any, List import logging class LLMMode(Enum): LEGACY = "legacy" BAML = "baml" HYBRID = "hybrid" # 双模式运行，用于验证 class LLMManager: def __init__(self, mode: LLMMode = LLMMode.HYBRID): self.mode = mode self.logger = logging.getLogger(__name__) # 初始化两种实现 if mode in [LLMMode.LEGACY, LLMMode.HYBRID]: from .legacy_impl import LegacyLLMClient self.legacy_client = LegacyLLMClient() if mode in [LLMMode.BAML, LLMMode.HYBRID]: from .baml_impl import BAMLLLMClient self.baml_client = BAMLLLMClient() async def process_request(self, agent_id: str, messages: List[Dict], context: Dict[str, Any] = None) -> Dict[str, Any]: context = context or {} if self.mode == LLMMode.LEGACY: return await self.legacy_client.process_request(agent_id, messages, context) elif self.mode == LLMMode.BAML: return await self.baml_client.process_request(agent_id, messages, context) elif self.mode == LLMMode.HYBRID: # 双模式运行，比较结果 legacy_result = await self.legacy_client.process_request(agent_id, messages, context) baml_result = await self.baml_client.process_request(agent_id, messages, context) # 记录差异用于验证 self._compare_results(legacy_result, baml_result, agent_id, context) # 默认返回BAML结果 return baml_result def _compare_results(self, legacy_result: Dict, baml_result: Dict, agent_id: str, context: Dict[str, Any]): """比较两种实现的结果差异""" # 实现详细的比较逻辑... pass

5. 高级特性与性能优化

5.1 流式响应处理

BAML原生支持流式响应，这对于需要实时显示LLM生成内容的场景非常重要：

// baml_src/functions/streaming_chat.baml function StreamingChatCompletion { input { messages: AgentMessage[] model: string stream: boolean = true } output stream { chunk: string is_final: boolean } impl<LLM> using model { prompt: """ {{#each messages}} <{{role}}>{{content}}</{{role}}> {{/each}} """ config { stream: stream } } }

对应的Python客户端代码：

# src/llm/baml_impl.py import asyncio from typing import AsyncGenerator from baml import BAMLClient class BAMLLLMClient: def __init__(self): self.client = BAMLClient() async def stream_chat(self, messages: List[Dict], model: str = None) -> AsyncGenerator[str, None]: """流式聊天完成""" async for chunk in self.client.StreamingChatCompletion( messages=messages, model=model or "openai-gpt4" ): yield chunk.content if chunk.is_final: break async def process_streaming_request(self, agent_id: str, messages: List[Dict], context: Dict[str, Any] = None) -> AsyncGenerator[Dict, None]: """处理流式智能体请求""" context = context or {} # 准备系统提示词 system_message = self._prepare_system_prompt(agent_id, context) all_messages = [system_message] + messages async for content_chunk in self.stream_chat(all_messages, context.get("model")): # 解析动作和内容 parsed_chunk = self._parse_streaming_chunk(content_chunk, context) yield parsed_chunk # 检查是否需要提前终止 if self._should_terminate_early(parsed_chunk): break

5.2 缓存策略与成本优化

利用BAML的缓存机制减少重复请求和降低成本：

# baml.config.yaml 中的缓存配置 caching: enabled: true strategy: "content_based" # 基于内容哈希的缓存 ttl: 3600 # 缓存1小时 storage: type: "redis" # 或 "memory", "file" config: host: "${REDIS_HOST}" port: 6379 database: 0 cost_tracking: enabled: true providers: - name: "openai" cost_per_token: prompt: 0.00003 # $0.03 per 1K tokens completion: 0.00006 # $0.06 per 1K tokens - name: "anthropic" cost_per_token: prompt: 0.000025 completion: 0.0001

5.3 监控与可观测性

集成监控系统，实时跟踪LLM性能和质量：

# src/monitoring/llm_metrics.py import time from dataclasses import dataclass from typing import Dict, Any import prometheus_client as prom from statsd import StatsClient @dataclass class LLMMetrics: request_count: prom.Counter = prom.Counter( 'llm_requests_total', 'Total LLM requests', ['model', 'status'] ) request_duration: prom.Histogram = prom.Histogram( 'llm_request_duration_seconds', 'LLM request duration', ['model'] ) token_usage: prom.Counter = prom.Counter( 'llm_tokens_total', 'Total tokens used', ['model', 'type'] ) class LLMMonitor: def __init__(self): self.metrics = LLMMetrics() self.statsd = StatsClient() def record_request(self, model: str, duration: float, tokens: Dict[str, int], success: bool): """记录LLM请求指标""" status = "success" if success else "error" # Prometheus指标 self.metrics.request_count.labels(model=model, status=status).inc() self.metrics.request_duration.labels(model=model).observe(duration) if tokens: self.metrics.token_usage.labels(model=model, type="prompt").inc(tokens.get('prompt_tokens', 0)) self.metrics.token_usage.labels(model=model, type="completion").inc(tokens.get('completion_tokens', 0)) # StatsD指标（用于实时告警） self.statsd.timing(f'llm.{model}.duration', duration * 1000) # 转毫秒 self.statsd.incr(f'llm.{model}.requests.{status}')

6. 测试策略与质量保障

6.1 BAML测试框架使用

BAML提供了强大的测试框架，可以编写针对LLM行为的测试用例：

// tests/baml_tests/agent_function_tests.baml test "AnalyzerAgent should provide structured analysis" { function: ProcessAgentRequest input: { request: { agent_id: "data-analyzer", session_id: "test-session-1", messages: [ { role: "user", content: "分析以下销售数据：Q1: 100万, Q2: 150万, Q3: 120万, Q4: 180万", timestamp: "2024-01-01T10:00:00Z" } ], context: { environment: "test", system_constraints: ["必须提供数据洞察", "需要给出建议"] } } } assert { // 检查响应是否成功 response.success == true, // 检查内容是否包含关键信息 response.content contains "增长", response.content contains "趋势", response.content contains "建议", // 检查元数据 response.metadata.tokens_used > 0, response.metadata.processing_time < 10.0 } } test "CoderAgent should generate valid code" { function: ProcessAgentRequest input: { request: { agent_id: "code-assistant", messages: [ { role: "user", content: "写一个Python函数计算斐波那契数列" } ], context: { environment: "test" } } } assert { response.success == true, response.content contains "def fibonacci", response.content contains "return", // 验证代码语法（通过外部验证器） validate_python_code(response.content) == true } }

6.2 集成测试与回归测试

建立完整的测试流水线，确保迁移不影响现有功能：

# tests/integration/test_llm_migration.py import pytest import asyncio from src.llm.llm_manager import LLMManager, LLMMode class TestLLMMigration: @pytest.fixture def test_messages(self): return [ {"role": "user", "content": "你好，请介绍一下你自己"}, {"role": "assistant", "content": "我是AI助手，很高兴为你服务"}, {"role": "user", "content": "今天的天气怎么样？"} ] @pytest.mark.asyncio async def test_hybrid_mode_consistency(self, test_messages): """测试双模式下结果的一致性""" manager = LLMManager(mode=LLMMode.HYBRID) # 相同的输入应该产生相似的结果 legacy_result = await manager.legacy_client.process_request( "test-agent", test_messages, {"environment": "test"} ) baml_result = await manager.baml_client.process_request( "test-agent", test_messages, {"environment": "test"} ) # 允许有一定的差异，但核心内容应该一致 assert legacy_result["success"] == baml_result["success"] assert len(legacy_result["content"]) > 0 assert len(baml_result["content"]) > 0 # 语义相似度检查（使用简单的文本相似度算法） similarity = self.calculate_similarity( legacy_result["content"], baml_result["content"] ) assert similarity > 0.7 # 70%相似度阈值 @pytest.mark.asyncio async def test_performance_comparison(self, test_messages): """性能对比测试""" import time manager = LLMManager(mode=LLMMode.HYBRID) # 测试旧版性能 start_time = time.time() legacy_result = await manager.legacy_client.process_request( "test-agent", test_messages, {"environment": "test"} ) legacy_duration = time.time() - start_time # 测试BAML性能 start_time = time.time() baml_result = await manager.baml_client.process_request( "test-agent", test_messages, {"environment": "test"} ) baml_duration = time.time() - start_time # BAML不应该比旧版慢太多（允许20%的性能差异） assert baml_duration <= legacy_duration * 1.2 print(f"旧版耗时: {legacy_duration:.3f}s, BAML耗时: {baml_duration:.3f}s")

6.3 质量门禁与自动化验证

建立质量门禁，确保只有通过测试的代码才能部署：

# .github/workflows/llm-migration-ci.yml name: LLM Migration CI on: push: branches: [ main, develop ] pull_request: branches: [ main ] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.9' - name: Install dependencies run: | pip install -r requirements.txt pip install -r requirements-dev.txt npm install -g @baml/cli - name: BAML schema validation run: | baml validate - name: Run BAML tests run: | baml test --verbose - name: Run integration tests run: | pytest tests/integration/ -v --cov=src/llm - name: Performance benchmark run: | python tests/performance/benchmark.py --threshold 1.2 - name: Security scan uses: anchore/scan-action@v3 with: image: myapp:latest quality-gate: needs: test runs-on: ubuntu-latest if: always() steps: - name: Check test results run: | # 检查所有测试是否通过 if [[ "${{ needs.test.result }}" != "success" ]]; then echo "❌ 测试失败，阻止合并" exit 1 fi echo "✅ 所有质量门禁通过"

7. 迁移后的架构优化与最佳实践

7.1 配置管理标准化

迁移完成后，建立统一的配置管理标准：

# src/config/llm_config.py from pydantic import BaseSettings, Field from typing import Dict, Any, Optional class LLMConfig(BaseSettings): """LLM配置模型""" # 模型提供商配置 openai_api_key: Optional[str] = Field(env="OPENAI_API_KEY") anthropic_api_key: Optional[str] = Field(env="ANTHROPIC_API_KEY") azure_openai_endpoint: Optional[str] = Field(env="AZURE_OPENAI_ENDPOINT") # 默认模型设置 default_model: str = "openai-gpt4" fallback_model: str = "anthropic-claude" # 性能配置 timeout: int = 30 max_retries: int = 3 circuit_breaker_threshold: int = 5 # 成本控制 monthly_budget: float = 1000.0 # 月度预算 cost_alert_threshold: float = 0.8 # 80%预算告警 class Config: env_file = ".env" case_sensitive = False class ModelConfig(BaseSettings): """单个模型配置""" name: str provider: str model: str temperature: float = 0.7 max_tokens: int = 1000 enabled: bool = True # 供应商特定配置 provider_config: Dict[str, Any] = {}

7.2 错误处理与重试策略优化

基于BAML的特性优化错误处理：

# src/llm/error_handling.py from typing import Type, Callable, Optional import asyncio import logging from dataclasses import dataclass from enum import Enum class LLMErrorType(Enum): RATE_LIMIT = "rate_limit" TIMEOUT = "timeout" AUTHENTICATION = "auth" CONTENT_FILTER = "content_filter" SERVER_ERROR = "server_error" NETWORK = "network" @dataclass class RetryConfig: max_attempts: int = 3 base_delay: float = 1.0 backoff_factor: float = 2.0 max_delay: float = 60.0 class LLMErrorHandler: def __init__(self, retry_config: RetryConfig = None): self.retry_config = retry_config or RetryConfig() self.logger = logging.getLogger(__name__) async def execute_with_retry(self, operation: Callable, error_mapping: Dict[Type[Exception], LLMErrorType] = None) -> Any: """带重试的执行操作""" last_exception = None for attempt in range(1, self.retry_config.max_attempts + 1): try: result = await operation() if attempt > 1: self.logger.info(f"操作在第 {attempt} 次尝试后成功") return result except Exception as e: last_exception = e error_type = self._classify_error(e, error_mapping) if not self._should_retry(error_type, attempt): raise delay = self._calculate_delay(attempt) self.logger.warning( f"LLM操作失败 (尝试 {attempt}/{self.retry_config.max_attempts}), " f"{error_type.value} 错误, {delay:.1f}秒后重试: {str(e)}" ) await asyncio.sleep(delay) # 所有重试都失败 self.logger.error( f"LLM操作在 {self.retry_config.max_attempts} 次尝试后仍失败: {str(last_exception)}" ) raise last_exception def _should_retry(self, error_type: LLMErrorType, attempt: int) -> bool: """判断是否应该重试""" non_retryable_errors = {LLMErrorType.AUTHENTICATION, LLMErrorType.CONTENT_FILTER} if error_type in non_retryable_errors: return False if attempt >= self.retry_config.max_attempts: return False return True

7.3 性能监控与优化建议

建立持续的性能监控和改进机制：

# src/optimization/llm_optimizer.py import asyncio from typing import List, Dict, Any from dataclasses import dataclass import statistics from datetime import datetime, timedelta @dataclass class PerformanceMetrics: p50_latency: float p95_latency: float success_rate: float tokens_per_second: float cost_per_request: float class LLMOptimizer: def __init__(self, metrics_collector): self.metrics_collector = metrics_collector self.optimization_rules = self._load_optimization_rules() async def analyze_performance(self, time_range: timedelta = timedelta(hours=1)) -> Dict[str, Any]: """分析LLM性能数据""" metrics_data = await self.metrics_collector.get_metrics(time_range) analysis = { "overall": self._calculate_overall_metrics(metrics_data), "by_model": self._calculate_model_metrics(metrics_data), "issues": self._identify_performance_issues(metrics_data), "recommendations": self._generate_recommendations(metrics_data) } return analysis def _generate_recommendations(self, metrics_data: List[Dict]) -> List[Dict]: """生成优化建议""" recommendations = [] # 基于规则生成建议 for rule in self.optimization_rules: if rule["condition"](metrics_data): recommendations.append({ "type": rule["type"], "priority": rule["priority"], "description": rule["description"], "action": rule["action"] }) return sorted(recommendations, key=lambda x: x