29.2 核心组件实现

约 494 字大约 2 分钟

29.2.1 LLM 客户端

LLM 客户端是编程 Agent 与大语言模型交互的核心组件，负责处理 API 调用、响应解析、错误处理等。

def init(self, config: LLMConfig): self.config = config self.api_key = config.api_key self.base_url = config.base_url self.model = config.model self.max_tokens = config.max_tokens self.temperature = config.temperature

会话管理

self.session = requests.Session() self.session.headers.update({ 'Authorization': f'Bearer {self.api_key}', 'Content-Type': 'application/json' })

缓存

self.cache = LRUCache(maxsize=1000)

统计

self.stats = { 'total_requests': 0, 'cache_hits': 0, 'errors': 0 }

async def complete(self, prompt: str, context: List[Dict] = None, **kwargs) -> str: """完成文本生成"""

检查缓存

cache_key = self._generate_cache_key(prompt, context, kwargs) cached_response = self.cache.get(cache_key) if cached_response: self.stats['cache_hits'] += 1 return cached_response

构建请求

messages = self._build_messages(prompt, context)

合并参数

params = { 'model': kwargs.get('model', self.model), 'messages': messages, 'max_tokens': kwargs.get('max_tokens', self.max_tokens), 'temperature': kwargs.get('temperature', self.temperature), **kwargs }

发送请求

try: response = await self._send_request(params) self.stats['total_requests'] += 1

解析响应

result = self._parse_response(response)

缓存结果

self.cache.set(cache_key, result)

return result

except Exception as e: self.stats['errors'] += 1

logger.error(f"LLM request failed: {e}") raise

async def _send_request(self, params: Dict) -> Dict: """发送请求""" url = f"{self.base_url}/chat/completions"

使用异步请求

async with aiohttp.ClientSession() as session: async with session.post(url, json=params) as response: if response.status != 200: error_text = await response.text() raise Exception(f"API error: {response.status} - {error_text}")

return await response.json()

def _build_messages(self, prompt: str, context: List[Dict] = None) -> List[Dict]: """构建消息列表""" messages = []

添加系统提示

if self.config.system_prompt: messages.append({ 'role': 'system', 'content': self.config.system_prompt })

添加上下文

if context: messages.extend(context)

添加用户提示

messages.append({ 'role': 'user', 'content': prompt })

return messages

def _parse_response(self, response: Dict) -> str: """解析响应""" try: return response['choices'][0]['message']['content'] except (KeyError, IndexError) as e: raise Exception(f"Invalid response format: {e}")

def _generate_cache_key(self, prompt: str, context: List[Dict], kwargs: Dict) -> str: """生成缓存键""" key_data = { 'prompt': prompt, 'context': context, 'kwargs': kwargs } return hashlib.md5( json.dumps(key_data, sort_keys=True).encode() ).hexdigest()

def get_stats(self) -> Dict[str, Any]: """获取统计信息""" return self.stats.copy()