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的结构化构建用户查询是传递具体任务需求的载体。一个结构化的查询能极大降低模型的误解概率。我通常将其分为三个部分任务指令清晰、无歧义地说明要做什么。使用祈使句如“生成以下函数的单元测试”、“重构这段代码提高其可读性”。上下文代码提供完整的、相关的代码片段。这包括目标函数/类本身以及可能重要的导入语句、类型定义或关键依赖。务必使用代码块格式提供。约束与细节列出所有具体要求。例如“使用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}‘, timeout5) 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(timeouttimeout) # 使用异步客户端提升性能 retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) 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”, jsonpayload, headersheaders ) 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“相关代码\npython\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 超时、重试与熔断机制网络服务不可靠我们必须为调用添加韧性。超时设置httpx客户端的timeout参数至关重要。它应包含连接超时、读超时和写超时。对于一个代码生成请求我通常设置总超时为30秒。如果模型响应慢超时后应快速失败而不是无限期等待。timeout_config httpx.Timeout(connect5.0, read25.0, write10.0, pool5.0) self.client httpx.AsyncClient(timeouttimeout_config)智能重试并非所有错误都值得重试。我们之前用retry装饰器处理了网络和5xx错误。但对于4xx客户端错误如401认证失败、400错误请求重试是无效的应该立即失败并报警。可以配置retry的retryretry_if_exception_type来细化重试条件。简易熔断器如果短时间内连续失败多次可能意味着下游服务不可用。可以实现一个简单的计数器在连续失败N次后暂时停止发送请求熔断经过一段冷却时间后再尝试恢复。这可以防止在服务宕机时你的应用还在疯狂重试浪费资源和时间。class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout60): 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 分层错误处理策略错误应该被分层捕获和处理从最具体的到最通用的API业务错误模型可能因为内容策略拒绝回答返回content_filter错误。需要在解析响应时检查choices[0].finish_reason。HTTP错误如429请求过多、503服务不可用。429错误应配合指数退避重试503错误可能需要更长的等待。网络错误连接超时、DNS解析失败等。这类错误适合重试。解析错误响应不是合法的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_idcall_id, modelself.model, input_token_estimateestimated_tokens) try: result await self.single_call(…) elapsed time.time() - start_time if result[“success”]: logger.info(“LLM call succeeded”, call_idcall_id, elapsed_secondsround(elapsed, 3), output_tokensresult[“usage”].get(“completion_tokens”), code_previewresult.get(“extracted_code”, “”)[:100] # 预览前100字符 ) else: logger.error(“LLM call failed”, call_idcall_id, errorresult[“error”], elapsed_secondsround(elapsed, 3)) return result except Exception as e: logger.exception(“Unexpected error in LLM call”, call_idcall_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, timeout5) - (bool, str): 在隔离环境中尝试执行代码适用于无依赖或简单依赖的脚本。 with tempfile.NamedTemporaryFile(mode‘w’, suffix‘.py’, deleteFalse) as f: f.write(code) temp_file_path f.name try: result subprocess.run([“python”, temp_file_path], capture_outputTrue, textTrue, timeouttimeout) 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-5分并标注问题所在如“逻辑错误”、“风格不符”、“缺少异常处理”。分析模式迭代提示词定期分析数据池。如果发现某一类任务如“生成数据库查询函数”的失败率或低分率很高就去审查对应的提示词。是不是约束条件没说清楚是不是提供的上下文不够然后有针对性地修改系统指令或用户查询的结构。例如通过分析发现模型生成的测试有时会忘记模拟mock外部依赖。那么在下一次迭代中就可以在系统指令中强化这一点“在编写测试时必须使用unittest.mock模块模拟所有网络请求和数据库调用。” 或者在用户查询的“约束”部分明确列出。这个“调用-验证-反馈-优化”的循环是让你的单次LLM调用从“能用”走向“好用”甚至“可靠”的关键。它让你不再是一个被动的API调用者而是一个主动的对话设计者和模型调优者。