1. 项目概述：从“一次调用”开始构建智能体

最近和几个做AI应用的朋友聊天，发现大家不约而同地都在折腾一个东西：Coding Agent（编程智能体）。无论是想自动生成单元测试、重构代码，还是想搞个能理解需求并直接输出可运行程序的“魔法”，最终都绕不开一个最基础、也最核心的环节——如何与大型语言模型（LLM）进行一次有效、可靠的对话。很多人一上来就想设计复杂的多轮对话、工具调用链，结果往往在第一步就卡住了：要么提示词（Prompt）写得不好，模型理解偏差；要么返回的代码格式混乱，无法直接使用；要么成本控制不住，调用一次就花掉不少预算。

这个项目，我们就从最原子的单元做起：实现一次高质量的LLM调用。这听起来简单，不就是发个HTTP请求，收个回复吗？但恰恰是这个“一次调用”，里面藏着构建稳定、高效Coding Agent的所有地基。这次调用，不仅要能准确理解我们的编程意图，还要返回结构清晰、可直接集成或执行的代码，同时兼顾响应速度、成本以及异常处理。我们将以构建一个“代码解释器”智能体的单次查询功能为例，拆解其中的每一个技术细节和设计考量。无论你是想入门AI编程，还是已经在构建复杂Agent体系，这次对基础单元的深度剖析，或许都能给你带来新的启发。

2. 核心设计：为编码任务量身定制对话

一次成功的LLM调用，其核心在于对话内容的设计。这远不止是简单地把问题丢给模型，而是需要精心构建一个“上下文环境”，让模型能扮演好“资深程序员”的角色。我们的目标是让模型完成一个具体任务，例如：“请为以下Python函数生成一个使用pytest框架的单元测试。” 为了实现这个目标，我们需要在单次调用中注入足够多的引导信息。

2.1 系统指令（System Prompt）的精准刻画

系统指令是定义模型角色和行为准则的关键。一个模糊的指令如“你是一个编程助手”，得到的结果可能参差不齐。我们必须进行精准刻画。

首先，明确核心身份。我会这样定义：“你是一个经验丰富的软件工程师，专注于编写高质量、可维护的代码。你精通多种编程语言，尤其擅长Python和JavaScript。你的回答必须务实、准确，直接提供解决方案。”

其次，设定输出规范。这是保证返回结果可直接使用的关键。指令中必须包含：

• 格式要求：明确要求代码必须包裹在标准的 Markdown 代码块中，并指定语言类型，例如```python。这便于后续自动化提取。
• 内容要求：禁止输出任何与代码无关的解释、开场白或总结。例如，不能说“当然，我很乐意为您生成测试代码，以下是示例：”，而应直接输出代码块。
• 假设与边界：要求模型在缺乏必要信息时，做出合理且安全的默认假设，并在代码中以注释说明。例如，若未指定异常处理方式，则默认进行日志记录而非直接忽略。

注意：不同的LLM提供商对系统指令的处理方式不同。例如，OpenAI的ChatCompletion API明确区分system和user角色，而一些开源模型可能将所有提示词都视为用户输入。在实际调用时，需要根据API规范进行调整。

2.2 用户查询（User Query）的结构化构建

用户查询是传递具体任务需求的载体。一个结构化的查询能极大降低模型的误解概率。我通常将其分为三个部分：

1. 任务指令：清晰、无歧义地说明要做什么。使用祈使句，如“生成以下函数的单元测试”、“重构这段代码，提高其可读性”。
2. 上下文代码：提供完整的、相关的代码片段。这包括目标函数/类本身，以及可能重要的导入语句、类型定义或关键依赖。务必使用代码块格式提供。
3. 约束与细节：列出所有具体要求。例如：
   • “使用pytest框架。”
   • “测试用例应覆盖正常输入、边界条件和异常输入。”
   • “模拟（mock）所有外部服务调用。”
   • “遵循PEP 8代码风格。”

一个完整的用户查询示例看起来是这样的：

请为以下Python函数生成单元测试，要求使用pytest框架，并模拟`requests.get`调用。 ```python import requests from typing import Optional def fetch_user_data(user_id: int) -> Optional[dict]: """根据用户ID从API获取用户数据。""" try: response = requests.get(f‘https://api.example.com/users/{user_id}‘, timeout=5) response.raise_for_status() return response.json() except (requests.RequestException, ValueError): return None

这种结构化的输入，让模型的任务聚焦度非常高，几乎能稳定地输出符合预期的测试代码。 ## 3. 技术实现：构建稳健的调用客户端 有了精心设计的提示词，下一步就是通过代码来实现调用。这里我们选择Python语言，因为它有丰富的生态。我们将构建一个不仅能够发送请求，还能处理各种边缘情况的稳健客户端。 ### 3.1 客户端封装与参数配置 我倾向于将LLM客户端封装成一个独立的类，这样便于管理配置、维护会话状态和处理错误。以下是一个基于OpenAI API（兼容OpenAI格式的各类开源模型网关）的核心实现框架： ```python import os import json from typing import Dict, Any, Optional, List import httpx from tenacity import retry, stop_after_attempt, wait_exponential class LiteLLMClient: def __init__(self, api_key: str, base_url: str = “https://api.openai.com/v1”, # 可替换为其他服务商地址 model: str = “gpt-4-turbo-preview”, timeout: int = 30, max_retries: int = 3): self.api_key = api_key self.base_url = base_url.rstrip(‘/’) self.model = model self.timeout = timeout self.max_retries = max_retries self.client = httpx.AsyncClient(timeout=timeout) # 使用异步客户端提升性能 @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) async def single_call(self, system_prompt: str, user_prompt: str, temperature: float = 0.2, max_tokens: Optional[int] = 2000) -> Dict[str, Any]: """执行单次LLM调用，并返回解析后的结果。""" messages = [ {“role”: “system”, “content”: system_prompt}, {“role”: “user”, “content”: user_prompt} ] payload = { “model”: self.model, “messages”: messages, “temperature”: temperature, “max_tokens”: max_tokens, “stream”: False # 单次调用，关闭流式传输以简化处理 } headers = { “Authorization”: f“Bearer {self.api_key}”, “Content-Type”: “application/json” } try: response = await self.client.post( f“{self.base_url}/chat/completions”, json=payload, headers=headers ) response.raise_for_status() result = response.json() # 基础解析 content = result[“choices”][0][“message”][“content”].strip() usage = result.get(“usage”, {}) return { “success”: True, “content”: content, “usage”: usage, “raw_response”: result # 保留原始响应以备排查 } except httpx.HTTPStatusError as e: # 处理HTTP错误（如429限速，5xx服务器错误） error_detail = f“HTTP error {e.response.status_code}: {e.response.text}” return {“success”: False, “error”: error_detail, “content”: “”} except (httpx.RequestError, json.JSONDecodeError, KeyError) as e: # 处理网络、解析和结构错误 return {“success”: False, “error”: f“Request or parsing error: {str(e)}”, “content”: “”}

关键参数解析：

• temperature (温度)：设置为0.2是一个经验值。对于编码任务，我们需要高度的确定性和一致性。温度越低（接近0），输出越确定、可重复；温度越高（接近1或2），输出越随机、有创造性。生成代码时，低温度能保证给定相同输入，输出稳定的代码。
• max_tokens (最大令牌数)：需要根据模型上下文窗口和任务复杂度估算。例如，GPT-4 Turbo上下文窗口为128K，但单次回复通常设置2000已足够生成一段完整的函数代码及其测试。设置过低会导致输出被截断，设置过高则浪费资源。
• retry (重试机制)：使用tenacity库实现指数退避重试，主要应对网络抖动和API的速率限制（429错误）。wait_exponential策略能在遇到临时性故障时自动重试，避免因偶发问题导致任务失败。

3.2 响应解析与后处理

模型返回的content是包含Markdown代码块的文本。我们的目标是精确提取出可执行的代码。一个健壮的解析器必不可少：

import re def extract_code_from_response(content: str, language: str = “python”) -> Optional[str]: """ 从LLM返回的文本中提取指定语言的代码块。 支持多个代码块，默认返回第一个匹配的。 """ # 匹配形如 ```python\ncode\n``` 的Markdown代码块 pattern = rf‘```{language}\s*(.*?)```‘ matches = re.findall(pattern, content, re.DOTALL) if matches: # 返回第一个代码块，并去除首尾空白 code = matches[0].strip() return code else: # 如果没有匹配到指定语言的代码块，尝试匹配任何代码块 fallback_pattern = r‘```(?:\w+)?\s*(.*?)```‘ fallback_matches = re.findall(fallback_pattern, content, re.DOTALL) if fallback_matches: # 这里可以记录一个警告日志：期望语言{language}，但实际未指定或为其他语言 return fallback_matches[0].strip() return None # 扩展客户端类，增加代码提取方法 class LiteLLMClient(LiteLLMClient): async def code_generation_call(self, task_description: str, context_code: str, constraints: List[str]) -> Dict[str, Any]: """专为代码生成任务封装的调用方法。""" system_prompt = “””你是一个资深软件工程师。请直接输出代码，不要任何解释。代码必须包裹在Markdown代码块中。如果需求不明确，做出合理的安全假设并用注释说明。“”” user_prompt_parts = [task_description] if context_code: user_prompt_parts.append(f“相关代码：\n```python\n{context_code}\n```”) if constraints: user_prompt_parts.append(“具体要求：” + “; “.join(constraints)) user_prompt = “\n\n”.join(user_prompt_parts) response = await self.single_call(system_prompt, user_prompt) if response[“success”]: extracted_code = extract_code_from_response(response[“content”]) response[“extracted_code”] = extracted_code else: response[“extracted_code”] = None return response

这个后处理流程确保了我们从模型的“自由发挥”中，精准地抓取出我们需要的、干净的代码片段，为后续的自动执行或集成到IDE铺平道路。

4. 成本控制与性能优化实战

在真实项目中，尤其是高频调用的场景下，成本和性能是必须严肃考虑的问题。一次调用看似微不足道，但积少成多。

4.1 令牌计算与成本估算

LLM API的计费通常基于输入和输出的令牌（Token）总数。我们需要有成本意识。以OpenAI的gpt-4-turbo-preview模型为例，其定价可能是输入$10/百万令牌，输出$30/百万令牌。

我们可以通过tiktoken库（OpenAI官方）或模型的Tokenizer进行近似估算，并在调用后记录实际消耗：

import tiktoken def estimate_tokens(text: str, model: str = “gpt-4”) -> int: """估算给定文本的令牌数。""" try: encoding = tiktoken.encoding_for_model(model) except KeyError: # 如果模型未找到，使用cl100k_base作为通用编码（GPT-3.5/4使用） encoding = tiktoken.get_encoding(“cl100k_base”) return len(encoding.encode(text)) # 在调用前估算 system_prompt_tokens = estimate_tokens(system_prompt) user_prompt_tokens = estimate_tokens(user_prompt) estimated_input_tokens = system_prompt_tokens + user_prompt_tokens print(f“预估输入令牌数：{estimated_input_tokens}”) # 调用后，从API响应中获取实际使用量 actual_input_tokens = response[“usage”].get(“prompt_tokens”, 0) actual_output_tokens = response[“usage”].get(“completion_tokens”, 0) total_cost = (actual_input_tokens / 1_000_000) * input_price_per_million + \ (actual_output_tokens / 1_000_000) * output_price_per_million print(f“本次调用成本：${total_cost:.6f}”)

实操心得：在系统指令中避免冗长的、与当前任务无关的通用描述。例如，不必每次都将“你是一个乐于助人的AI助手”这种话写进去。可以将其作为客户端默认配置的一部分，仅在初始化时加载一次，而不是放在每次调用的消息里。这能有效减少重复的令牌消耗。

4.2 超时、重试与熔断机制

网络服务不可靠，我们必须为调用添加韧性。

1. 超时设置：httpx客户端的timeout参数至关重要。它应包含连接超时、读超时和写超时。对于一个代码生成请求，我通常设置总超时为30秒。如果模型响应慢，超时后应快速失败，而不是无限期等待。

   timeout_config = httpx.Timeout(connect=5.0, read=25.0, write=10.0, pool=5.0) self.client = httpx.AsyncClient(timeout=timeout_config)

2. 智能重试：并非所有错误都值得重试。我们之前用@retry装饰器处理了网络和5xx错误。但对于4xx客户端错误（如401认证失败、400错误请求），重试是无效的，应该立即失败并报警。可以配置retry的retry=retry_if_exception_type来细化重试条件。

3. 简易熔断器：如果短时间内连续失败多次，可能意味着下游服务不可用。可以实现一个简单的计数器，在连续失败N次后，暂时停止发送请求（熔断），经过一段冷却时间后再尝试恢复。这可以防止在服务宕机时，你的应用还在疯狂重试，浪费资源和时间。

class CircuitBreaker: def __init__(self, failure_threshold=5, recovery_timeout=60): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.failure_count = 0 self.last_failure_time = None self.state = “CLOSED” # CLOSED, OPEN, HALF-OPEN def call_allowed(self): if self.state == “OPEN”: if time.time() - self.last_failure_time > self.recovery_timeout: self.state = “HALF-OPEN” # 进入半开状态尝试恢复 return True return False return True def record_success(self): self.failure_count = 0 if self.state == “HALF-OPEN”: self.state = “CLOSED” def record_failure(self): self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.state = “OPEN”

将熔断器集成到客户端中，在每次调用前检查call_allowed()，调用成功后record_success()，失败后record_failure()。

5. 错误处理与日志记录体系

一次健壮的调用必须能妥善处理所有异常，并留下清晰的日志，方便问题追踪和调试。

5.1 分层错误处理策略

错误应该被分层捕获和处理，从最具体的到最通用的：

1. API业务错误：模型可能因为内容策略拒绝回答，返回content_filter错误。需要在解析响应时检查choices[0].finish_reason。
2. HTTP错误：如429（请求过多）、503（服务不可用）。429错误应配合指数退避重试；503错误可能需要更长的等待。
3. 网络错误：连接超时、DNS解析失败等。这类错误适合重试。
4. 解析错误：响应不是合法的JSON，或者JSON结构不符合预期。这类错误通常意味着API端点或版本可能发生了变化，需要人工介入检查。

在我们的single_call方法中，已经通过多个except块进行了分层处理。但我们可以做得更细致，例如，将不同的错误类型映射到不同的重试策略或报警级别。

5.2 结构化日志记录

使用如structlog或logging模块记录结构化日志，这对于后续分析调用模式、排查问题、计算成本至关重要。每一条日志应包含：

• 请求ID：唯一标识一次调用，串联起请求和响应。
• 时间戳。
• 模型名称。
• 输入/输出令牌数估算值与实际值。
• 耗时。
• 最终状态：成功/失败。
• 错误类型（如果失败）。
• 提取的代码片段的前N个字符（脱敏后，用于快速验证结果）。

import logging import uuid import time logger = logging.getLogger(__name__) async def single_call_with_logging(self, …): call_id = str(uuid.uuid4())[:8] start_time = time.time() logger.info(“LLM call started”, call_id=call_id, model=self.model, input_token_estimate=estimated_tokens) try: result = await self.single_call(…) elapsed = time.time() - start_time if result[“success”]: logger.info(“LLM call succeeded”, call_id=call_id, elapsed_seconds=round(elapsed, 3), output_tokens=result[“usage”].get(“completion_tokens”), code_preview=result.get(“extracted_code”, “”)[:100] # 预览前100字符 ) else: logger.error(“LLM call failed”, call_id=call_id, error=result[“error”], elapsed_seconds=round(elapsed, 3)) return result except Exception as e: logger.exception(“Unexpected error in LLM call”, call_id=call_id) return {“success”: False, “error”: f“Unexpected error: {str(e)}”, “content”: “”}

这样的日志，当你在凌晨三点被报警叫醒时，能让你快速定位到是哪个请求、因为什么原因失败了，而不是面对一片“调用出错”的模糊信息。

6. 效果评估与迭代改进

完成一次调用并拿到代码后，工作只完成了一半。我们如何知道这次调用是“好”的？我们需要一个评估和反馈闭环。

6.1 建立自动化验证基线

对于代码生成任务，最直接的验证就是代码能否通过语法检查。我们可以集成ast（抽象语法树）模块进行快速验证：

import ast import tempfile import subprocess def validate_python_syntax(code: str) -> (bool, str): """验证Python代码的语法是否正确。""" try: ast.parse(code) return True, “Syntax OK” except SyntaxError as e: return False, f“Syntax error: {e.msg} at line {e.lineno}” def validate_code_execution(code: str, timeout=5) -> (bool, str): """在隔离环境中尝试执行代码（适用于无依赖或简单依赖的脚本）。""" with tempfile.NamedTemporaryFile(mode=‘w’, suffix=‘.py’, delete=False) as f: f.write(code) temp_file_path = f.name try: result = subprocess.run([“python”, temp_file_path], capture_output=True, text=True, timeout=timeout) if result.returncode == 0: return True, “Execution succeeded” else: return False, f“Execution failed: {result.stderr}” except subprocess.TimeoutExpired: return False, “Execution timeout” finally: import os os.unlink(temp_file_path)

对于单元测试生成这类任务，验证可以更进一步：自动运行生成的测试，看它是否能通过（当然，这需要能访问到被测试的原始代码）。通过设置这样的自动化检查，我们可以立即过滤掉那些连语法都不通的糟糕输出。

6.2 构建反馈数据池与提示词迭代

每次调用，无论成功与否，都是一次学习机会。我建议建立一个简单的反馈机制：

1. 存储输入输出对：将每次调用的系统指令、用户查询、模型原始响应、提取的代码、验证结果、令牌用量和成本，存储到数据库或文件中。这构成了你的专属数据池。
2. 人工标注与评分：对于重要的任务，引入人工审核环节。让开发人员对生成的代码质量进行评分（例如，1-5分），并标注问题所在（如“逻辑错误”、“风格不符”、“缺少异常处理”）。
3. 分析模式，迭代提示词：定期分析数据池。如果发现某一类任务（如“生成数据库查询函数”）的失败率或低分率很高，就去审查对应的提示词。是不是约束条件没说清楚？是不是提供的上下文不够？然后有针对性地修改系统指令或用户查询的结构。

例如，通过分析发现，模型生成的测试有时会忘记模拟（mock）外部依赖。那么在下一次迭代中，就可以在系统指令中强化这一点：“在编写测试时，必须使用unittest.mock模块模拟所有网络请求和数据库调用。” 或者，在用户查询的“约束”部分明确列出。

这个“调用-验证-反馈-优化”的循环，是让你的单次LLM调用从“能用”走向“好用”甚至“可靠”的关键。它让你不再是一个被动的API调用者，而是一个主动的对话设计者和模型调优者。