深入解析LLM请求响应循环：从令牌化到流式输出的完整技术流程

当你点击发送按钮，向ChatGPT提出一个问题时，背后发生了什么？为什么有时候响应飞快，有时候却要等待几秒钟？为什么同样的提示词在不同时间会得到不同质量的回答？

这些问题背后，隐藏着一个看似简单实则复杂的技术过程：LLM（大语言模型）的请求与响应循环。理解这个过程，不仅能帮你更好地使用AI工具，还能在开发AI应用时避免常见的性能瓶颈和错误处理问题。

很多人以为LLM就是"输入问题，输出答案"的黑箱，但实际上，从你的键盘敲击到屏幕显示结果，中间经历了令牌化、上下文管理、推理计算、流式输出等多个关键环节。每个环节都可能成为影响最终体验的关键因素。

本文将深入拆解LLM请求响应的完整生命周期，通过实际代码示例展示每个阶段的工作原理，并分享在生产环境中优化这一流程的实用技巧。无论你是AI应用开发者还是技术爱好者，都能从中获得可落地的技术洞察。

1. 为什么需要深入理解LLM请求响应循环？

在AI应用开发中，单纯会调用API是远远不够的。当你的应用从demo走向生产环境时，会遇到各种意想不到的问题：为什么响应时间波动这么大？为什么有时会收到截断的回复？如何有效处理长文档问答？

理解LLM请求响应循环的真正价值在于：

性能优化：知道在哪个环节可以节省时间成本。比如，令牌化过程通常很快，但模型推理时间与输出长度直接相关。

成本控制：LLM API通常按令牌收费。了解令牌计数机制可以帮助你设计更经济的提示词结构。

错误处理：能够准确诊断问题是出在网络层、认证层还是模型层，而不是简单地"重试一下"。

用户体验：实现流式输出、合理设置超时时间、处理部分失败情况，都需要对底层机制有清晰认识。

更重要的是，随着AI应用架构的复杂化，从简单的问答场景发展到多步骤推理、工具调用、智能体协作等高级模式，对基础流程的深入理解变得尤为关键。

2. LLM请求响应循环的核心组件

在深入流程之前，我们需要先了解参与这个循环的关键组件及其职责。

2.1 客户端应用程序

这是用户直接交互的界面，可以是Web应用、移动App或命令行工具。客户端负责收集用户输入、格式化请求、处理认证、发送API调用，并最终向用户展示结果。

2.2 API网关与负载均衡器

大型LLM服务提供商通常有分布式的服务器集群。API网关负责请求路由、速率限制、认证验证，并将请求分发到合适的后端服务器。

2.3 令牌化器（Tokenizer）

这是LLM流程中的第一个关键组件，负责将自然语言文本转换为模型能够理解的数字序列（令牌ID）。不同的模型使用不同的令牌化方案，这直接影响文本处理的效率和成本。

2.4 LLM推理引擎

核心的模型推理组件，接收令牌序列，基于预训练权重进行前向传播计算，生成下一个令牌的概率分布。现代推理引擎通常经过高度优化，支持批处理、持续批处理等加速技术。

2.5 解码策略控制器

决定如何从模型的概率分布中选择下一个令牌。常见的策略包括贪婪解码、束搜索、核采样等，不同的策略会在生成质量和多样性之间做出权衡。

2.6 上下文管理器

负责维护对话历史或文档上下文，确保模型在生成回复时能够参考之前的交互内容。上下文长度限制是这一组件的主要约束条件。

3. 请求准备阶段：从用户输入到API调用

当用户提交一个问题时，真正的LLM流程其实才刚刚开始。

3.1 提示词构建与格式化

不同的模型对输入格式有特定要求。以ChatGPT为例，需要将对话历史转换为特定的消息格式：

def build_chatml_prompt(messages): """ 构建OpenAI ChatML格式的提示词 """ formatted_messages = [] for msg in messages: if msg['role'] == 'system': formatted_messages.append(f"<|im_start|>system\n{msg['content']}<|im_end|>") elif msg['role'] == 'user': formatted_messages.append(f"<|im_start|>user\n{msg['content']}<|im_end|>") elif msg['role'] == 'assistant': formatted_messages.append(f"<|im_start|>assistant\n{msg['content']}<|im_end|>") # 添加当前助理回复的开始标记 formatted_messages.append("<|im_start|>assistant\n") return "\n".join(formatted_messages) # 示例使用 messages = [ {"role": "system", "content": "你是一个有用的助手。"}, {"role": "user", "content": "请解释机器学习的基本概念。"} ] prompt = build_chatml_prompt(messages) print(prompt)

这种格式化的目的是明确区分不同角色的发言，帮助模型理解对话结构。

3.2 令牌化过程详解

令牌化是将文本转换为模型可理解数字序列的关键步骤。让我们通过实际代码理解这个过程：

# 使用Hugging Face transformers库演示令牌化过程 from transformers import AutoTokenizer # 加载GPT-2的令牌化器（与GPT系列类似） tokenizer = AutoTokenizer.from_pretrained("gpt2") def demonstrate_tokenization(text): print(f"原始文本: {text}") print(f"文本长度: {len(text)} 字符") # 令牌化 tokens = tokenizer.tokenize(text) token_ids = tokenizer.encode(text) print(f"令牌列表: {tokens}") print(f"令牌数量: {len(tokens)}") print(f"令牌ID序列: {token_ids}") # 反向令牌化验证 decoded_text = tokenizer.decode(token_ids) print(f"解码后文本: {decoded_text}") print("-" * 50) # 测试不同文本的令牌化效果 test_texts = [ "Hello, world!", "机器学习很有趣。", "The quick brown fox jumps over the lazy dog.", "这是一个测试文本，用于演示令牌化过程。" ] for text in test_texts: demonstrate_tokenization(text)

运行这个示例，你会发现英文文本通常一个单词可能被分成多个令牌，而中文文本往往每个汉字对应一个令牌。这种差异直接影响API调用成本和处理效率。

3.3 API请求构建

构建完整的API请求需要包含认证信息、模型参数和提示词内容：

import requests import json def build_llm_request(api_key, model, messages, temperature=0.7, max_tokens=1000): """ 构建标准的LLM API请求 """ headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": False # 非流式响应 } return headers, payload # 实际调用示例 def call_openai_api(api_key, messages, model="gpt-3.5-turbo"): headers, payload = build_llm_request(api_key, model, messages) try: response = requests.post( "https://api.openai.com/v1/chat/completions", headers=headers, json=payload, timeout=30 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"API调用失败: {e}") return None # 使用示例 api_key = "your-api-key-here" # 实际使用时替换为真实API密钥 messages = [ {"role": "user", "content": "请用简单的话解释人工智能是什么？"} ] # 实际调用（注释掉以避免意外收费） # result = call_openai_api(api_key, messages) # if result: # print(result['choices'][0]['message']['content'])

4. 服务器端处理流程

当API请求到达服务器后，真正的模型推理过程开始。

4.1 请求验证与预处理

服务器首先验证API密钥、检查速率限制、确认模型可用性。然后对输入文本进行令牌化，并检查是否超出模型上下文限制。

def validate_request(input_text, model_context_limit=4096): """ 模拟服务器端的请求验证过程 """ # 令牌化并检查长度 token_count = len(tokenizer.encode(input_text)) if token_count > model_context_limit: return False, f"输入过长: {token_count}令牌，限制: {model_context_limit}" # 检查内容安全策略（简化版） blocked_terms = ["恶意内容", "敏感信息"] # 示例屏蔽词 for term in blocked_terms: if term in input_text: return False, "内容违反安全策略" return True, f"验证通过，令牌数: {token_count}" # 测试验证逻辑 test_input = "请帮我写一段关于机器学习的介绍文字。" is_valid, message = validate_request(test_input) print(f"验证结果: {is_valid}, 信息: {message}")

4.2 模型推理与文本生成

这是最核心的环节，模型基于输入序列生成输出令牌。以下代码模拟了简化的生成过程：

import numpy as np def simulate_text_generation(input_ids, vocab_size=50257, max_new_tokens=50): """ 简化版的文本生成模拟 实际LLM使用复杂的神经网络计算 """ generated_ids = input_ids.copy() for i in range(max_new_tokens): # 模拟模型计算下一个令牌的概率分布 # 实际中这是通过Transformer前向传播计算的 logits = np.random.randn(vocab_size) # 模拟logits输出 probabilities = softmax(logits) # 使用温度采样选择下一个令牌 temperature = 0.8 scaled_logits = logits / temperature scaled_probs = softmax(scaled_logits) next_token = np.random.choice(vocab_size, p=scaled_probs) generated_ids.append(next_token) # 模拟遇到停止符的情况 if next_token == tokenizer.eos_token_id: break return generated_ids def softmax(x): """Softmax函数实现""" exp_x = np.exp(x - np.max(x)) return exp_x / np.sum(exp_x) # 模拟生成过程 input_text = "人工智能是" input_ids = tokenizer.encode(input_text) generated_ids = simulate_text_generation(input_ids) generated_text = tokenizer.decode(generated_ids) print(f"输入: {input_text}") print(f"生成结果: {generated_text}")

4.3 流式输出处理

流式输出可以显著改善用户体验，让用户逐步看到生成结果：

import time def simulate_streaming_generation(prompt, chunk_size=3): """ 模拟流式文本生成 """ full_response = "大型语言模型是通过大量文本数据训练的深度学习模型，能够理解和生成人类语言。" # 模拟逐词生成 words = full_response.split() accumulated_response = "" for i in range(0, len(words), chunk_size): chunk = words[i:i + chunk_size] new_text = " ".join(chunk) # 如果不是第一个chunk，需要添加空格 if accumulated_response: new_text = " " + new_text accumulated_response += new_text yield accumulated_response # 模拟生成延迟 time.sleep(0.5) # 测试流式输出 prompt = "请解释什么是大型语言模型？" print("用户提问:", prompt) print("模型回复:", end=" ") for partial_response in simulate_streaming_generation(prompt): # 模拟清行效果（实际中可能使用更复杂的终端控制） print("\r" + " " * 100, end="") # 清空当前行 print(f"\r模型回复: {partial_response}", end="", flush=True) print() # 最终换行

5. 响应返回与客户端处理

服务器生成完成后，响应需要经过处理后返回给客户端。

5.1 响应格式标准化

不同的LLM提供商可能使用不同的响应格式，但通常包含相似的信息：

def format_api_response(completion_text, prompt_tokens, completion_tokens, model): """ 标准化API响应格式 """ response = { "id": f"chatcmpl-{np.random.randint(100000, 999999)}", # 模拟ID生成 "object": "chat.completion", "created": int(time.time()), "model": model, "choices": [ { "index": 0, "message": { "role": "assistant", "content": completion_text }, "finish_reason": "stop" # 或 "length", "content_filter" } ], "usage": { "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": prompt_tokens + completion_tokens } } return response # 示例响应 example_response = format_api_response( "大型语言模型是...", prompt_tokens=50, completion_tokens=100, model="gpt-3.5-turbo" ) print(json.dumps(example_response, indent=2, ensure_ascii=False))

5.2 客户端响应处理

客户端需要处理各种可能的响应情况，包括成功响应、错误、速率限制等：

def handle_api_response(response): """ 综合处理API响应 """ if response is None: return "请求失败，请检查网络连接" if response.status_code == 200: data = response.json() return data['choices'][0]['message']['content'] elif response.status_code == 429: return "请求过于频繁，请稍后重试" elif response.status_code == 401: return "API密钥无效，请检查认证信息" elif response.status_code == 400: error_info = response.json() if "maximum context length" in error_info.get('error', {}).get('message', ''): return "输入文本过长，请缩短后重试" else: return f"请求参数错误: {error_info}" elif response.status_code == 503: return "服务暂时不可用，请稍后重试" else: return f"未知错误: {response.status_code}" # 模拟各种响应场景 class MockResponse: def __init__(self, status_code, json_data=None): self.status_code = status_code self._json_data = json_data or {} def json(self): return self._json_data # 测试不同的响应处理 test_cases = [ MockResponse(200, {"choices": [{"message": {"content": "这是一个测试回复"}}]}), MockResponse(429), MockResponse(401), MockResponse(400, {"error": {"message": "This model's maximum context length is 4096 tokens"}}), MockResponse(503) ] for i, case in enumerate(test_cases): result = handle_api_response(case) print(f"测试案例 {i+1}: {result}")

6. 完整工作流程实战演示

现在让我们整合所有环节，构建一个完整的LLM对话系统：

import threading import queue import time class SimpleLLMClient: """ 简化的LLM客户端实现 演示完整的请求响应循环 """ def __init__(self, api_key, model="gpt-3.5-turbo", max_history=10): self.api_key = api_key self.model = model self.conversation_history = [] self.max_history = max_history def add_message(self, role, content): """添加消息到对话历史""" self.conversation_history.append({"role": role, "content": content}) # 保持历史记录不超过限制 if len(self.conversation_history) > self.max_history * 2: # 用户和助理消息各算一条 self.conversation_history = self.conversation_history[-self.max_history*2:] def send_message(self, user_message, temperature=0.7, max_tokens=500): """发送消息并获取回复""" self.add_message("user", user_message) # 模拟API调用延迟 print("思考中", end="", flush=True) for i in range(3): time.sleep(0.5) print(".", end="", flush=True) print() # 模拟模型回复（实际中调用真实API） simulated_response = self._simulate_llm_response(user_message) self.add_message("assistant", simulated_response) return simulated_response def _simulate_llm_response(self, user_message): """模拟LLM生成回复""" # 基于用户消息的关键词生成简单回复 responses = { "介绍": "我是一个大型语言模型，能够理解和生成自然语言文本。", "帮助": "我可以回答问题、生成文本、翻译语言、总结内容等。", "天气": "我无法获取实时天气信息，建议查看专业天气服务。", "时间": f"当前模拟时间是：{time.strftime('%Y-%m-%d %H:%M:%S')}", } for keyword, response in responses.items(): if keyword in user_message: return response return "这是一个模拟回复。在实际应用中，这里会是真实的模型生成内容。" def stream_message(self, user_message, callback): """流式消息发送（模拟）""" self.add_message("user", user_message) def generate_stream(): response_text = "这是一个流式回复演示。模型会逐步生成文本内容。" words = response_text.split() for i in range(len(words)): partial = " ".join(words[:i+1]) callback(partial) time.sleep(0.3) self.add_message("assistant", response_text) # 在后台线程中生成流式响应 thread = threading.Thread(target=generate_stream) thread.start() def get_conversation_summary(self): """获取对话摘要""" total_chars = sum(len(msg["content"]) for msg in self.conversation_history) return f"对话轮次: {len(self.conversation_history)//2}, 总字符数: {total_chars}" # 使用示例 def demo_complete_workflow(): client = SimpleLLMClient("simulated-api-key") # 模拟对话 messages = [ "请介绍一下你自己", "你能做什么？", "今天的天气怎么样？", "现在几点钟？" ] for msg in messages: print(f"用户: {msg}") response = client.send_message(msg) print(f"助理: {response}") print("-" * 40) print("流式输出演示:") def stream_callback(partial_text): print(f"\r助理: {partial_text}", end="", flush=True) client.stream_message("请用流式方式回复", stream_callback) time.sleep(5) # 等待流式输出完成 print(f"\n对话统计: {client.get_conversation_summary()}") # 运行演示 demo_complete_workflow()

7. 性能优化与最佳实践

在实际生产环境中，优化LLM请求响应循环可以显著提升用户体验并降低成本。

7.1 提示词工程优化

有效的提示词设计可以减少不必要的令牌使用：

def optimize_prompt(original_prompt): """ 提示词优化示例 """ optimization_tips = { "避免冗余": "删除不必要的礼貌用语和重复内容", "明确指令": "使用清晰的指令词，如'总结'、'列表'、'解释'", "提供示例": "对于复杂任务，提供输入输出示例", "指定格式": "明确要求输出格式，如JSON、Markdown等", "分步思考": "对于复杂问题，要求模型逐步推理" } optimized = original_prompt # 删除常见冗余短语 redundant_phrases = [ "请问你可以", "能不能请你", "我希望你", "非常感谢你的帮助", "如果你不介意的话" ] for phrase in redundant_phrases: optimized = optimized.replace(phrase, "") # 确保以动词开头 if not any(optimized.startswith(verb) for verb in ["总结", "解释", "分析", "比较", "生成"]): # 添加明确的指令 optimized = "请回答: " + optimized return optimized # 测试提示词优化 test_prompts = [ "请问你可以帮我解释一下机器学习的概念吗？非常感谢！", "能不能请你总结一下这篇文章的主要内容？希望不要太长。", "如果你不介意的话，我想了解一下人工智能的发展历史。" ] for prompt in test_prompts: optimized = optimize_prompt(prompt) print(f"原始: {prompt}") print(f"优化: {optimized}") print()

7.2 缓存策略实现

对于重复或相似的查询，使用缓存可以大幅减少API调用：

import hashlib from functools import lru_cache class PromptCache: """ 提示词缓存系统 """ def __init__(self, max_size=1000): self.cache = {} self.max_size = max_size def _get_hash(self, text): """生成文本哈希作为缓存键""" return hashlib.md5(text.encode()).hexdigest() def get(self, prompt): """获取缓存结果""" key = self._get_hash(prompt) return self.cache.get(key) def set(self, prompt, response): """设置缓存结果""" if len(self.cache) >= self.max_size: # 简单的LRU策略：移除第一个项目 first_key = next(iter(self.cache)) del self.cache[first_key] key = self._get_hash(prompt) self.cache[key] = { 'response': response, 'timestamp': time.time() } def get_with_cache(self, prompt, api_call_function): """带缓存的API调用""" cached = self.get(prompt) if cached: print("使用缓存结果") return cached['response'] print("调用API") response = api_call_function(prompt) self.set(prompt, response) return response # 使用缓存示例 cache = PromptCache() def simulated_api_call(prompt): """模拟API调用""" time.sleep(1) # 模拟网络延迟 return f"响应: {prompt}" # 测试缓存效果 test_prompt = "什么是人工智能？" print("第一次调用（应调用API）:") start_time = time.time() result1 = cache.get_with_cache(test_prompt, simulated_api_call) print(f"结果: {result1}, 耗时: {time.time() - start_time:.2f}秒") print("\n第二次调用（应使用缓存）:") start_time = time.time() result2 = cache.get_with_cache(test_prompt, simulated_api_call) print(f"结果: {result2}, 耗时: {time.time() - start_time:.2f}秒")

7.3 错误处理与重试机制

健壮的错误处理是生产环境应用的必备特性：

import random from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class RobustLLMClient: """ 具有重试机制的健壮LLM客户端 """ def __init__(self, api_key, max_retries=3): self.api_key = api_key self.max_retries = max_retries @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), retry=retry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)) ) def call_api_with_retry(self, payload): """带重试机制的API调用""" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } # 模拟各种错误场景 error_scenario = random.choice([ "success", "timeout", "connection_error", "rate_limit" ]) if error_scenario == "success": return {"choices": [{"message": {"content": "成功响应"}}]} elif error_scenario == "timeout": raise requests.exceptions.Timeout("请求超时") elif error_scenario == "connection_error": raise requests.exceptions.ConnectionError("连接错误") elif error_scenario == "rate_limit": response = MockResponse(429) raise requests.exceptions.HTTPError("速率限制", response=response) def safe_api_call(self, prompt): """安全的API调用，包含完整的错误处理""" for attempt in range(self.max_retries + 1): try: response = self.call_api_with_retry(prompt) return response except requests.exceptions.Timeout: print(f"尝试 {attempt + 1}: 请求超时") if attempt == self.max_retries: return {"error": "请求超时，请稍后重试"} except requests.exceptions.ConnectionError: print(f"尝试 {attempt + 1}: 连接错误") if attempt == self.max_retries: return {"error": "网络连接问题，请检查网络"} except requests.exceptions.HTTPError as e: if e.response.status_code == 429: wait_time = 2 ** attempt # 指数退避 print(f"尝试 {attempt + 1}: 速率限制，等待{wait_time}秒") time.sleep(wait_time) else: return {"error": f"HTTP错误: {e.response.status_code}"} except Exception as e: return {"error": f"未知错误: {str(e)}"} return {"error": "达到最大重试次数"} # 测试错误处理 client = RobustLLMClient("test-key") result = client.safe_api_call("测试提示词") print(f"最终结果: {result}")

8. 常见问题与解决方案

在实际使用LLM API时，开发者经常会遇到一些典型问题。以下是常见问题及其解决方案：

8.1 令牌超限问题

问题现象：收到"maximum context length"错误，或者回复被意外截断。

解决方案：

def handle_context_overflow(text, model_max_tokens=4096, reserve_tokens=100): """ 处理上下文超限问题 """ tokens = tokenizer.encode(text) if len(tokens) <= model_max_tokens - reserve_tokens: return text # 无需处理 # 计算需要保留的令牌数 max_allowed = model_max_tokens - reserve_tokens # 策略1: 直接截断（简单但可能丢失重要信息） truncated_tokens = tokens[:max_allowed] truncated_text = tokenizer.decode(truncated_tokens) # 策略2: 智能截断（优先保留开头和结尾） if len(tokens) > max_allowed: # 保留开头和结尾，删除中间部分 keep_start = max_allowed // 3 keep_end = max_allowed - keep_start start_tokens = tokens[:keep_start] end_tokens = tokens[-keep_end:] smart_tokens = start_tokens + [tokenizer.sep_token_id] + end_tokens smart_text = tokenizer.decode(smart_tokens) else: smart_text = text return { 'original_length': len(tokens), 'truncated_text': truncated_text, 'smart_truncated_text': smart_text, 'recommendation': '使用智能截断保留重要信息' if len(tokens) > max_allowed else '文本长度合适' } # 测试长文本处理 long_text = "这是一段很长的文本。" * 1000 # 创建长文本 result = handle_context_overflow(long_text, model_max_tokens=100, reserve_tokens=20) print(f"原始令牌数: {result['original_length']}") print(f"截断后长度: {len(tokenizer.encode(result['truncated_text']))}") print(f"推荐方案: {result['recommendation']}")

8.2 响应质量不稳定

问题现象：相同提示词得到质量波动很大的回复。

解决方案：

• 使用更低的temperature值（如0.3-0.5）获得更一致的输出
• 提供更明确的指令和示例
• 使用系统消息设定模型行为边界
• 对重要任务实施多次生成并选择最佳结果

8.3 速率限制处理

问题现象：收到429状态码，API调用被限制。

解决方案：

class RateLimitManager: """ 速率限制管理 """ def __init__(self, requests_per_minute=60): self.requests_per_minute = requests_per_minute self.request_times = [] def can_make_request(self): """检查是否可以发起新请求""" current_time = time.time() # 移除1分钟前的记录 self.request_times = [t for t in self.request_times if current_time - t < 60] return len(self.request_times) < self.requests_per_minute def record_request(self): """记录请求时间""" self.request_times.append(time.time()) def wait_if_needed(self): """如果需要等待，则阻塞直到可以发起请求""" while not self.can_make_request(): oldest_time = min(self.request_times) wait_time = 60 - (time.time() - oldest_time) if wait_time > 0: print(f"达到速率限制，等待{wait_time:.1f}秒") time.sleep(wait_time) self.record_request() # 使用示例 rate_limiter = RateLimitManager(requests_per_minute=3) # 测试用低限制 for i in range(5): rate_limiter.wait_if_needed() print(f"发起请求 {i+1} at {time.strftime('%H:%M:%S')}") time.sleep(1) # 模拟请求处理

9. 高级特性与未来展望

随着LLM技术的发展，请求响应循环也在不断进化，涌现出许多高级特性。

9.1 函数调用能力

现代LLM支持函数调用，允许模型请求执行外部工具：

def demonstrate_function_calling(): """ 演示函数调用模式 """ functions = [ { "name": "get_weather", "description": "获取指定城市的天气信息", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "城市名称" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位" } }, "required": ["location"] } } ] # 模拟模型决定调用函数 model_response = { "role": "assistant", "content": None, "function_call": { "name": "get_weather", "arguments": '{"location": "北京", "unit": "celsius"}' } } print("模型请求调用函数:", model_response['function_call']['name']) print("参数:", model_response['function_call']['arguments']) # 执行函数 function_name = model_response['function_call']['name'] arguments = json.loads(model_response['function_call']['arguments']) if function_name == "get_weather": # 模拟天气查询 weather_info = { "location": arguments['location'], "temperature": 25, "unit": arguments['unit'], "condition": "晴朗" } # 将函数结果返回给模型继续处理 follow_up_messages = [ {"role": "function", "name": "get_weather", "content": json.dumps(weather_info)} ] print("函数执行结果已返回给模型继续处理") demonstrate_function_calling()

9.2 视觉模型集成

多模态模型支持图像和文本的联合处理：

def demonstrate_multimodal_capability(): """ 演示多模态处理能力 """ multimodal_prompt = [ { "role": "user", "content": [ {"type": "text", "text": "请描述这张图片中的内容"}, {"type": "image_url", "image_url": {"url": "https://example.com/image.jpg"}} ] } ] print("多模态请求示例:") for content in multimodal_prompt[0]["content"]: if content["type"] == "text": print(f"文本内容: {content['text']}") else: print(f"图像URL: {content['image_url']['url']}") # 模拟模型处理流程 processing_steps = [ "图像编码与特征提取", "文本令牌化", "多模态融合处理", "文本生成" ] print("\n处理步骤:") for step in processing_steps: print(f"- {step}") demonstrate_multimodal_capability()

9.3 自我批判与修正

高级模型能够进行自我评估和修正：

def demonstrate_self_correction(): """ 演示自我修正能力 """ initial_response = "太阳围绕地球旋转。" # 错误陈述 critique_prompt = f""" 请对以下陈述进行批判性评估，如果发现错误请提供修正: 陈述: {initial_response} 请按以下格式回复: 1. 评估: [正确/部分正确/错误] 2. 解释: [详细解释] 3. 修正: [修正后的陈述] """ # 模拟批判过程 critique_response = """ 1. 评估: 错误 2. 解释: 实际上地球围绕太阳旋转，这是日心说的基本观点。太阳是太阳系的中心天体。 3. 修正: 地球围绕太阳旋转。 """ print("初始陈述:", initial_response) print("自我批判结果:") print(critique_response) demonstrate_self_correction()

理解LLM请求响应循环的完整流程，是开发现代AI应用的基础。从简单的文本补全到复杂的多模态推理，这个基本循环在不断演进中支撑着越来越强大的AI能力。

随着模型技术的进步，我们可以期待更高效的令牌化方案、更智能的上下文管理、更可靠的质量控制机制。但无论技术如何发展，对基础流程的深入理解始终是有效利用这些工具的关键。

建议在实际项目中从简单用例开始，逐步增加复杂度，并在每个阶段密切关注性能指标和用户体验。通过持续的测试和优化，你将能够构建出既强大又可靠的AI应用系统。