在实际 LLM 应用开发中如何高效、稳定地向模型输入内容直接影响着交互体验、系统性能和最终输出质量。无论是构建一个简单的问答机器人还是一个复杂的多模态智能体输入环节的设计都远不止“用户打字模型回复”这么简单。它涉及到输入格式的规范化、上下文的组织、长文本的处理、多模态数据的融合以及面对不同服务提供商 API 的兼容性挑战。本文将围绕 LLM 输入这一核心环节从基础的文本构造方法讲起逐步深入到文件处理、上下文管理、多模态集成以及生产环境下的稳定性保障。我们会通过具体的代码示例、配置片段和排查路径说明如何构建一个健壮、可扩展的 LLM 输入管道。1. 理解 LLM 输入的基本结构从消息列表到提示词模板LLM 的输入并非简单的字符串而是一个结构化的消息列表。理解这个结构是有效交互的基础。1.1 消息角色系统user、assistant 与 system主流的 LLM API如 OpenAI、Anthropic通常采用基于角色的消息系统。每条消息都包含一个role字段和一个content字段。# 一个典型的对话输入结构 messages [ {role: system, content: 你是一个专业的软件工程师助手用中文回答技术问题。}, {role: user, content: 如何在 Python 中优雅地处理 JSON 解析异常}, {role: assistant, content: 可以使用 try-except 块捕获 json.JSONDecodeError 异常。}, {role: user, content: 能给我一个具体的代码示例吗} ]三种核心角色及其作用system: 设定助手的身份、行为准则和回复风格。这条消息通常位于对话开头对整个会话起到“定向”作用。user: 代表终端用户的问题或指令。这是需要模型回应的主要输入。assistant: 模型之前的回复记录。在多轮对话中它提供了对话历史和上下文。注意不是所有模型都严格区分这三种角色。某些开源模型或特定 API 可能使用简化的输入格式。在实际集成前务必查阅对应模型的文档。1.2 提示词模板将动态变量注入固定结构直接拼接字符串容易出错且难以维护。提示词模板Prompt Template通过占位符的方式将可变部分与固定结构分离。from string import Template # 使用 Python 标准库的字符串模板 qa_template Template( 请基于以下上下文回答问题。 上下文$context 问题$question 请用中文回答如果上下文中没有答案请明确说明“根据已知信息无法回答”。 ) # 填充模板 context LLM 的输入通常是一个消息列表包含 system、user、assistant 三种角色。 question LLM 输入中有哪几种角色 prompt qa_template.substitute(contextcontext, questionquestion) print(prompt)在实际项目中更常使用 LangChain 等框架提供的模板工具它们支持更复杂的变量处理和类型检查。from langchain.prompts import PromptTemplate # 使用 LangChain 的提示词模板 template 你是一个技术支持专家。请根据提供的文档片段回答问题。 文档内容 {document} 用户问题{question} 请用简洁的技术语言回答 prompt PromptTemplate( input_variables[document, question], templatetemplate ) # 填充模板 formatted_prompt prompt.format( documentAPI 密钥需要保存在环境变量中不要硬编码在代码里。, question如何安全地存储 API 密钥 )1.3 输入长度管理与令牌计数LLM 对输入长度有限制即上下文窗口。在构造输入时必须考虑令牌数量。# 使用 tiktoken 库估算 OpenAI 模型的令牌数 import tiktoken def count_tokens(text, modelgpt-3.5-turbo): encoding tiktoken.encoding_for_model(model) return len(encoding.encode(text)) message 这是一个测试文本用于计算令牌数量。 token_count count_tokens(message) print(f文本的令牌数量: {token_count}) # 检查是否超过模型限制 max_tokens 4096 # gpt-3.5-turbo 的上下文窗口 if token_count max_tokens: print(输入过长需要截断或压缩)对于长文档常见的处理策略包括截断直接截取前 N 个令牌。滑动窗口将文档分块每次只输入相关部分。摘要压缩先用模型对长文档进行摘要再将摘要作为上下文。2. 处理文件输入PDF、Word 与文本提取很多场景需要将整个文件内容输入给 LLM如文档问答、合同分析等。关键是准确提取文件中的文本信息。2.1 PDF 文件文本提取PDF 解析的准确性直接影响后续的 LLM 处理效果。# 使用 PyPDF2 提取文本 import PyPDF2 def extract_text_from_pdf(pdf_path): text try: with open(pdf_path, rb) as file: reader PyPDF2.PdfReader(file) for page in reader.pages: text page.extract_text() \n except Exception as e: print(fPDF 解析错误: {e}) return None return text # 使用 pdfplumber通常更准确 import pdfplumber def extract_text_with_pdfplumber(pdf_path): text try: with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: text page.extract_text() or # 处理无文本页面 except Exception as e: print(fPDF 解析错误: {e}) return None return text # 提取示例 pdf_text extract_text_with_pdfplumber(technical_spec.pdf) if pdf_text: # 清理文本移除多余换行和空格 cleaned_text .join(pdf_text.split()) print(f提取文本长度: {len(cleaned_text)} 字符)2.2 Word 文档处理# 使用 python-docx 处理 .docx 文件 from docx import Document def extract_text_from_docx(docx_path): try: doc Document(docx_path) text [] for paragraph in doc.paragraphs: if paragraph.text.strip(): # 跳过空段落 text.append(paragraph.text) return \n.join(text) except Exception as e: print(fWord 文档解析错误: {e}) return None2.3 文本预处理与分块直接输入长文档可能超出模型限制需要合理的分块策略。from langchain.text_splitter import RecursiveCharacterTextSplitter def chunk_text(text, chunk_size1000, chunk_overlap200): 将长文本分割成重叠的小块 splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen ) chunks splitter.split_text(text) return chunks # 示例处理提取的 PDF 文本 if pdf_text: chunks chunk_text(pdf_text) print(f将文档分割为 {len(chunks)} 个块) # 可以选择最相关的块输入给 LLM first_chunk chunks[0] print(f第一个块的内容: {first_chunk[:200]}...)3. 构建对话上下文管理器多轮对话需要维护历史记录但又要避免无限增长导致超出上下文窗口。3.1 基础对话历史管理class ConversationManager: def __init__(self, system_message, max_tokens4000): self.messages [] self.max_tokens max_tokens self.current_tokens 0 if system_message: self.add_message(system, system_message) def add_message(self, role, content): 添加新消息并更新令牌计数 message {role: role, content: content} self.messages.append(message) # 简化令牌计数使用字符数近似估计实际项目应用更准确的方法 self.current_tokens len(content) // 4 # 近似估算 # 如果超出限制移除最早的用户/助手对话对 while self.current_tokens self.max_tokens and len(self.messages) 1: # 保留 system 消息移除最早的非 system 消息 if len(self.messages) 1 and self.messages[1][role] ! system: removed self.messages.pop(1) self.current_tokens - len(removed[content]) // 4 def get_messages(self): return self.messages.copy() def clear_history(self): 清空对话历史但保留系统消息 if self.messages and self.messages[0][role] system: system_msg self.messages[0] self.messages [system_msg] self.current_tokens len(system_msg[content]) // 4 else: self.messages [] self.current_tokens 0 # 使用示例 manager ConversationManager( system_message你是一个编程助手用简洁的技术语言回答。, max_tokens2000 # 预留空间给回复 ) manager.add_message(user, 如何用 Python 读取 CSV 文件) manager.add_message(assistant, 可以使用 csv 模块或 pandas 库。) manager.add_message(user, 能展示 pandas 的示例吗) print(当前对话消息:, len(manager.get_messages()))3.2 基于重要性的上下文窗口优化简单的先进先出策略可能会丢失重要信息。更智能的方法包括总结压缩将旧对话总结成更短的版本重要性评分根据消息类型、关键词等保留重要消息主题分段按主题分割对话只保留当前主题的历史def summarize_conversation(messages, model_client): 使用 LLM 对长对话进行摘要 if len(messages) 4: # 对话不长时不需要摘要 return messages # 构建摘要提示词 summary_prompt 请将以下对话总结成一个简洁的段落保留主要的技术讨论点和结论。 对话记录 for msg in messages: summary_prompt f\n{msg[role]}: {msg[content]} summary_prompt \n\n摘要 # 调用模型进行摘要简化示例 try: # 这里是调用 LLM 的伪代码 # summary model_client.chat.completions.create(...) # 返回摘要 最近几条消息 recent_messages messages[-2:] # 保留最近两条 summarized [{role: system, content: 先前对话已总结 摘要内容...}] return summarized recent_messages except Exception as e: print(f对话摘要失败: {e}) return messages[-4:] # 失败时回退到保留最近4条4. 多模态输入处理图像、表格与结构化数据现代 LLM 逐渐支持多模态输入如图像理解、表格数据分析等。4.1 图像输入处理对于支持视觉的模型如 GPT-4V需要将图像转换为 base64 编码或提供 URL。import base64 import requests def image_to_base64(image_path): 将图像文件转换为 base64 字符串 with open(image_path, rb) as image_file: return base64.b64encode(image_file.read()).decode(utf-8) def prepare_multimodal_message(text, image_pathNone): 准备包含文本和图像的消息 message_content [{type: text, text: text}] if image_path: base64_image image_to_base64(image_path) message_content.append({ type: image_url, image_url: { url: fdata:image/jpeg;base64,{base64_image} } }) return [{role: user, content: message_content}] # 使用示例 multimodal_message prepare_multimodal_message( 请描述这张图片中的内容, screenshot.png )4.2 表格数据处理对于结构化数据可以选择转换为文本描述或保留表格格式。import pandas as pd def dataframe_to_text_description(df, max_rows10): 将 DataFrame 转换为文本描述避免直接输入过大表格 if len(df) max_rows: description f表格包含 {len(df)} 行 {len(df.columns)} 列数据。前{max_rows}行示例\n sample_df df.head(max_rows) else: description 表格数据\n sample_df df # 转换为标记格式的文本 description sample_df.to_markdown(indexFalse) if len(df) max_rows: description f\n... 还有 {len(df) - max_rows} 行数据 return description # 示例处理 CSV 数据 df pd.read_csv(sales_data.csv) table_description dataframe_to_text_description(df) messages [ {role: user, content: f请分析以下销售数据\n{table_description}} ]5. 生产环境下的输入管道与错误处理在实际生产环境中输入处理需要更强的鲁棒性和可观测性。5.1 输入验证与清理import re def validate_and_clean_input(text, max_length10000): 验证和清理用户输入 if not text or not isinstance(text, str): raise ValueError(输入必须是非空字符串) if len(text) max_length: raise ValueError(f输入长度超过限制 {max_length} 字符) # 清理潜在的恶意字符或异常格式 cleaned re.sub(r[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F], , text) # 移除控制字符 cleaned cleaned.strip() if len(cleaned) 2: raise ValueError(输入内容过短) return cleaned def safe_prepare_messages(user_input, conversation_historyNone, system_promptNone): 安全地准备消息列表 try: # 验证输入 cleaned_input validate_and_clean_input(user_input) messages [] # 添加系统提示词 if system_prompt: messages.append({role: system, content: system_prompt}) # 添加对话历史 if conversation_history: messages.extend(conversation_history) # 添加当前用户输入 messages.append({role: user, content: cleaned_input}) return messages except ValueError as e: print(f输入验证失败: {e}) return None except Exception as e: print(f消息准备异常: {e}) return None5.2 重试机制与故障转移import time from typing import List, Dict, Any class RobustLLMClient: def __init__(self, primary_client, fallback_clientNone): self.primary_client primary_client self.fallback_client fallback_client self.max_retries 3 self.retry_delay 1 def send_request(self, messages: List[Dict], model: str, **kwargs) - Any: 带重试和故障转移的 LLM 请求 last_exception None # 重试主客户端 for attempt in range(self.max_retries): try: response self.primary_client.chat.completions.create( modelmodel, messagesmessages, **kwargs ) return response except Exception as e: last_exception e print(f主客户端请求失败 (尝试 {attempt 1}/{self.max_retries}): {e}) if attempt self.max_retries - 1: time.sleep(self.retry_delay * (2 ** attempt)) # 指数退避 # 主客户端全部失败尝试备用客户端 if self.fallback_client: try: print(尝试备用客户端...) response self.fallback_client.chat.completions.create( modelmodel, messagesmessages, **kwargs ) return response except Exception as e: last_exception e print(f备用客户端也失败: {e}) raise last_exception if last_exception else Exception(所有请求尝试失败)5.3 输入输出日志与监控生产环境需要完整的可观测性。import logging import json from datetime import datetime class LLMInteractionLogger: def __init__(self, log_filellm_interactions.log): self.logger logging.getLogger(LLMInteraction) handler logging.FileHandler(log_file) formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) handler.setFormatter(formatter) self.logger.addHandler(handler) self.logger.setLevel(logging.INFO) def log_interaction(self, messages, response, model, duration, successTrue): 记录完整的交互信息 log_entry { timestamp: datetime.now().isoformat(), model: model, duration_seconds: duration, success: success, input_messages: messages, output: response.choices[0].message.content if success else str(response) } # 敏感信息脱敏 sanitized_entry self.sanitize_log_entry(log_entry) self.logger.info(json.dumps(sanitized_entry, ensure_asciiFalse)) def sanitize_log_entry(self, entry): 脱敏处理移除或混淆敏感信息 sanitized entry.copy() # 示例隐藏可能的 API 密钥片段 if input_messages in sanitized: for msg in sanitized[input_messages]: if content in msg: msg[content] re.sub(rsk-\w{20}, sk-***, msg[content]) return sanitized # 使用示例 logger LLMInteractionLogger() def monitored_llm_call(messages, model, client): start_time time.time() try: response client.chat.completions.create( modelmodel, messagesmessages ) duration time.time() - start_time logger.log_interaction(messages, response, model, duration, successTrue) return response except Exception as e: duration time.time() - start_time logger.log_interaction(messages, e, model, duration, successFalse) raise6. 常见问题排查与优化建议在实际项目中LLM 输入环节经常会遇到各种问题。下面是一些典型场景的排查路径。6.1 输入相关错误排查表问题现象可能原因检查方式处理建议模型返回无关内容系统提示词未生效或被覆盖检查消息列表中 system 角色消息的位置和内容确保 system 消息是第一条内容清晰明确对话上下文丢失超出上下文窗口限制计算消息总令牌数检查模型限制实现对话历史管理移除最早消息或进行摘要模型拒绝回答输入格式错误或包含敏感词检查消息结构验证内容规范性确保消息角色正确清理输入内容处理长文档时性能差单次输入过长或分块策略不合理监控令牌使用量分析分块大小优化文本分块策略优先输入相关段落多模态输入失败图像格式不支持或编码错误验证图像格式、大小和编码方式转换为模型支持的格式检查 base64 编码6.2 输入管道性能优化# 异步处理提高吞吐量 import asyncio from concurrent.futures import ThreadPoolExecutor class AsyncInputProcessor: def __init__(self, max_workers5): self.executor ThreadPoolExecutor(max_workersmax_workers) async def process_input_batch(self, inputs): 批量处理输入如图片转 base64、文本分块等 loop asyncio.get_event_loop() # 将同步的 CPU 密集型任务放到线程池中执行 tasks [ loop.run_in_executor(self.executor, self.process_single_input, input_data) for input_data in inputs ] results await asyncio.gather(*tasks, return_exceptionsTrue) return results def process_single_input(self, input_data): 处理单个输入同步方法 # 这里可以是文本清理、文件解析等操作 if isinstance(input_data, str): return validate_and_clean_input(input_data) else: # 处理文件或其他类型输入 return input_data # 使用示例 async def main(): processor AsyncInputProcessor() inputs [输入1, 输入2, 输入3] results await processor.process_input_batch(inputs) print(f处理完成 {len(results)} 个输入) # asyncio.run(main())6.3 安全最佳实践输入验证层层把关前端验证防止明显恶意输入后端验证严格的内容检查和长度限制模型前验证最终的内容清理和格式化敏感信息过滤def filter_sensitive_info(text): 过滤可能的敏感信息 patterns [ r\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b, # 信用卡号 r\b\d{3}[- ]?\d{2}[- ]?\d{4}\b, # SSN rsk-\w{20,}, # API 密钥 ] for pattern in patterns: text re.sub(pattern, [FILTERED], text) return text速率限制与滥用防护实现用户级别的请求频率限制监控异常输入模式设置合理的超时时间构建可靠的 LLM 输入管道需要综合考虑格式规范、内容处理、上下文管理和生产环境要求。从准确的文件解析到智能的对话管理从多模态支持到健壮的错误处理每个环节都直接影响最终的用户体验和系统稳定性。在实际项目中建议从最小可行方案开始逐步迭代加入更高级的功能和保障机制。
LLM输入管道构建：从消息结构到生产级对话管理实践
发布时间：2026/7/25 16:26:24
在实际 LLM 应用开发中如何高效、稳定地向模型输入内容直接影响着交互体验、系统性能和最终输出质量。无论是构建一个简单的问答机器人还是一个复杂的多模态智能体输入环节的设计都远不止“用户打字模型回复”这么简单。它涉及到输入格式的规范化、上下文的组织、长文本的处理、多模态数据的融合以及面对不同服务提供商 API 的兼容性挑战。本文将围绕 LLM 输入这一核心环节从基础的文本构造方法讲起逐步深入到文件处理、上下文管理、多模态集成以及生产环境下的稳定性保障。我们会通过具体的代码示例、配置片段和排查路径说明如何构建一个健壮、可扩展的 LLM 输入管道。1. 理解 LLM 输入的基本结构从消息列表到提示词模板LLM 的输入并非简单的字符串而是一个结构化的消息列表。理解这个结构是有效交互的基础。1.1 消息角色系统user、assistant 与 system主流的 LLM API如 OpenAI、Anthropic通常采用基于角色的消息系统。每条消息都包含一个role字段和一个content字段。# 一个典型的对话输入结构 messages [ {role: system, content: 你是一个专业的软件工程师助手用中文回答技术问题。}, {role: user, content: 如何在 Python 中优雅地处理 JSON 解析异常}, {role: assistant, content: 可以使用 try-except 块捕获 json.JSONDecodeError 异常。}, {role: user, content: 能给我一个具体的代码示例吗} ]三种核心角色及其作用system: 设定助手的身份、行为准则和回复风格。这条消息通常位于对话开头对整个会话起到“定向”作用。user: 代表终端用户的问题或指令。这是需要模型回应的主要输入。assistant: 模型之前的回复记录。在多轮对话中它提供了对话历史和上下文。注意不是所有模型都严格区分这三种角色。某些开源模型或特定 API 可能使用简化的输入格式。在实际集成前务必查阅对应模型的文档。1.2 提示词模板将动态变量注入固定结构直接拼接字符串容易出错且难以维护。提示词模板Prompt Template通过占位符的方式将可变部分与固定结构分离。from string import Template # 使用 Python 标准库的字符串模板 qa_template Template( 请基于以下上下文回答问题。 上下文$context 问题$question 请用中文回答如果上下文中没有答案请明确说明“根据已知信息无法回答”。 ) # 填充模板 context LLM 的输入通常是一个消息列表包含 system、user、assistant 三种角色。 question LLM 输入中有哪几种角色 prompt qa_template.substitute(contextcontext, questionquestion) print(prompt)在实际项目中更常使用 LangChain 等框架提供的模板工具它们支持更复杂的变量处理和类型检查。from langchain.prompts import PromptTemplate # 使用 LangChain 的提示词模板 template 你是一个技术支持专家。请根据提供的文档片段回答问题。 文档内容 {document} 用户问题{question} 请用简洁的技术语言回答 prompt PromptTemplate( input_variables[document, question], templatetemplate ) # 填充模板 formatted_prompt prompt.format( documentAPI 密钥需要保存在环境变量中不要硬编码在代码里。, question如何安全地存储 API 密钥 )1.3 输入长度管理与令牌计数LLM 对输入长度有限制即上下文窗口。在构造输入时必须考虑令牌数量。# 使用 tiktoken 库估算 OpenAI 模型的令牌数 import tiktoken def count_tokens(text, modelgpt-3.5-turbo): encoding tiktoken.encoding_for_model(model) return len(encoding.encode(text)) message 这是一个测试文本用于计算令牌数量。 token_count count_tokens(message) print(f文本的令牌数量: {token_count}) # 检查是否超过模型限制 max_tokens 4096 # gpt-3.5-turbo 的上下文窗口 if token_count max_tokens: print(输入过长需要截断或压缩)对于长文档常见的处理策略包括截断直接截取前 N 个令牌。滑动窗口将文档分块每次只输入相关部分。摘要压缩先用模型对长文档进行摘要再将摘要作为上下文。2. 处理文件输入PDF、Word 与文本提取很多场景需要将整个文件内容输入给 LLM如文档问答、合同分析等。关键是准确提取文件中的文本信息。2.1 PDF 文件文本提取PDF 解析的准确性直接影响后续的 LLM 处理效果。# 使用 PyPDF2 提取文本 import PyPDF2 def extract_text_from_pdf(pdf_path): text try: with open(pdf_path, rb) as file: reader PyPDF2.PdfReader(file) for page in reader.pages: text page.extract_text() \n except Exception as e: print(fPDF 解析错误: {e}) return None return text # 使用 pdfplumber通常更准确 import pdfplumber def extract_text_with_pdfplumber(pdf_path): text try: with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: text page.extract_text() or # 处理无文本页面 except Exception as e: print(fPDF 解析错误: {e}) return None return text # 提取示例 pdf_text extract_text_with_pdfplumber(technical_spec.pdf) if pdf_text: # 清理文本移除多余换行和空格 cleaned_text .join(pdf_text.split()) print(f提取文本长度: {len(cleaned_text)} 字符)2.2 Word 文档处理# 使用 python-docx 处理 .docx 文件 from docx import Document def extract_text_from_docx(docx_path): try: doc Document(docx_path) text [] for paragraph in doc.paragraphs: if paragraph.text.strip(): # 跳过空段落 text.append(paragraph.text) return \n.join(text) except Exception as e: print(fWord 文档解析错误: {e}) return None2.3 文本预处理与分块直接输入长文档可能超出模型限制需要合理的分块策略。from langchain.text_splitter import RecursiveCharacterTextSplitter def chunk_text(text, chunk_size1000, chunk_overlap200): 将长文本分割成重叠的小块 splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen ) chunks splitter.split_text(text) return chunks # 示例处理提取的 PDF 文本 if pdf_text: chunks chunk_text(pdf_text) print(f将文档分割为 {len(chunks)} 个块) # 可以选择最相关的块输入给 LLM first_chunk chunks[0] print(f第一个块的内容: {first_chunk[:200]}...)3. 构建对话上下文管理器多轮对话需要维护历史记录但又要避免无限增长导致超出上下文窗口。3.1 基础对话历史管理class ConversationManager: def __init__(self, system_message, max_tokens4000): self.messages [] self.max_tokens max_tokens self.current_tokens 0 if system_message: self.add_message(system, system_message) def add_message(self, role, content): 添加新消息并更新令牌计数 message {role: role, content: content} self.messages.append(message) # 简化令牌计数使用字符数近似估计实际项目应用更准确的方法 self.current_tokens len(content) // 4 # 近似估算 # 如果超出限制移除最早的用户/助手对话对 while self.current_tokens self.max_tokens and len(self.messages) 1: # 保留 system 消息移除最早的非 system 消息 if len(self.messages) 1 and self.messages[1][role] ! system: removed self.messages.pop(1) self.current_tokens - len(removed[content]) // 4 def get_messages(self): return self.messages.copy() def clear_history(self): 清空对话历史但保留系统消息 if self.messages and self.messages[0][role] system: system_msg self.messages[0] self.messages [system_msg] self.current_tokens len(system_msg[content]) // 4 else: self.messages [] self.current_tokens 0 # 使用示例 manager ConversationManager( system_message你是一个编程助手用简洁的技术语言回答。, max_tokens2000 # 预留空间给回复 ) manager.add_message(user, 如何用 Python 读取 CSV 文件) manager.add_message(assistant, 可以使用 csv 模块或 pandas 库。) manager.add_message(user, 能展示 pandas 的示例吗) print(当前对话消息:, len(manager.get_messages()))3.2 基于重要性的上下文窗口优化简单的先进先出策略可能会丢失重要信息。更智能的方法包括总结压缩将旧对话总结成更短的版本重要性评分根据消息类型、关键词等保留重要消息主题分段按主题分割对话只保留当前主题的历史def summarize_conversation(messages, model_client): 使用 LLM 对长对话进行摘要 if len(messages) 4: # 对话不长时不需要摘要 return messages # 构建摘要提示词 summary_prompt 请将以下对话总结成一个简洁的段落保留主要的技术讨论点和结论。 对话记录 for msg in messages: summary_prompt f\n{msg[role]}: {msg[content]} summary_prompt \n\n摘要 # 调用模型进行摘要简化示例 try: # 这里是调用 LLM 的伪代码 # summary model_client.chat.completions.create(...) # 返回摘要 最近几条消息 recent_messages messages[-2:] # 保留最近两条 summarized [{role: system, content: 先前对话已总结 摘要内容...}] return summarized recent_messages except Exception as e: print(f对话摘要失败: {e}) return messages[-4:] # 失败时回退到保留最近4条4. 多模态输入处理图像、表格与结构化数据现代 LLM 逐渐支持多模态输入如图像理解、表格数据分析等。4.1 图像输入处理对于支持视觉的模型如 GPT-4V需要将图像转换为 base64 编码或提供 URL。import base64 import requests def image_to_base64(image_path): 将图像文件转换为 base64 字符串 with open(image_path, rb) as image_file: return base64.b64encode(image_file.read()).decode(utf-8) def prepare_multimodal_message(text, image_pathNone): 准备包含文本和图像的消息 message_content [{type: text, text: text}] if image_path: base64_image image_to_base64(image_path) message_content.append({ type: image_url, image_url: { url: fdata:image/jpeg;base64,{base64_image} } }) return [{role: user, content: message_content}] # 使用示例 multimodal_message prepare_multimodal_message( 请描述这张图片中的内容, screenshot.png )4.2 表格数据处理对于结构化数据可以选择转换为文本描述或保留表格格式。import pandas as pd def dataframe_to_text_description(df, max_rows10): 将 DataFrame 转换为文本描述避免直接输入过大表格 if len(df) max_rows: description f表格包含 {len(df)} 行 {len(df.columns)} 列数据。前{max_rows}行示例\n sample_df df.head(max_rows) else: description 表格数据\n sample_df df # 转换为标记格式的文本 description sample_df.to_markdown(indexFalse) if len(df) max_rows: description f\n... 还有 {len(df) - max_rows} 行数据 return description # 示例处理 CSV 数据 df pd.read_csv(sales_data.csv) table_description dataframe_to_text_description(df) messages [ {role: user, content: f请分析以下销售数据\n{table_description}} ]5. 生产环境下的输入管道与错误处理在实际生产环境中输入处理需要更强的鲁棒性和可观测性。5.1 输入验证与清理import re def validate_and_clean_input(text, max_length10000): 验证和清理用户输入 if not text or not isinstance(text, str): raise ValueError(输入必须是非空字符串) if len(text) max_length: raise ValueError(f输入长度超过限制 {max_length} 字符) # 清理潜在的恶意字符或异常格式 cleaned re.sub(r[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F], , text) # 移除控制字符 cleaned cleaned.strip() if len(cleaned) 2: raise ValueError(输入内容过短) return cleaned def safe_prepare_messages(user_input, conversation_historyNone, system_promptNone): 安全地准备消息列表 try: # 验证输入 cleaned_input validate_and_clean_input(user_input) messages [] # 添加系统提示词 if system_prompt: messages.append({role: system, content: system_prompt}) # 添加对话历史 if conversation_history: messages.extend(conversation_history) # 添加当前用户输入 messages.append({role: user, content: cleaned_input}) return messages except ValueError as e: print(f输入验证失败: {e}) return None except Exception as e: print(f消息准备异常: {e}) return None5.2 重试机制与故障转移import time from typing import List, Dict, Any class RobustLLMClient: def __init__(self, primary_client, fallback_clientNone): self.primary_client primary_client self.fallback_client fallback_client self.max_retries 3 self.retry_delay 1 def send_request(self, messages: List[Dict], model: str, **kwargs) - Any: 带重试和故障转移的 LLM 请求 last_exception None # 重试主客户端 for attempt in range(self.max_retries): try: response self.primary_client.chat.completions.create( modelmodel, messagesmessages, **kwargs ) return response except Exception as e: last_exception e print(f主客户端请求失败 (尝试 {attempt 1}/{self.max_retries}): {e}) if attempt self.max_retries - 1: time.sleep(self.retry_delay * (2 ** attempt)) # 指数退避 # 主客户端全部失败尝试备用客户端 if self.fallback_client: try: print(尝试备用客户端...) response self.fallback_client.chat.completions.create( modelmodel, messagesmessages, **kwargs ) return response except Exception as e: last_exception e print(f备用客户端也失败: {e}) raise last_exception if last_exception else Exception(所有请求尝试失败)5.3 输入输出日志与监控生产环境需要完整的可观测性。import logging import json from datetime import datetime class LLMInteractionLogger: def __init__(self, log_filellm_interactions.log): self.logger logging.getLogger(LLMInteraction) handler logging.FileHandler(log_file) formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) handler.setFormatter(formatter) self.logger.addHandler(handler) self.logger.setLevel(logging.INFO) def log_interaction(self, messages, response, model, duration, successTrue): 记录完整的交互信息 log_entry { timestamp: datetime.now().isoformat(), model: model, duration_seconds: duration, success: success, input_messages: messages, output: response.choices[0].message.content if success else str(response) } # 敏感信息脱敏 sanitized_entry self.sanitize_log_entry(log_entry) self.logger.info(json.dumps(sanitized_entry, ensure_asciiFalse)) def sanitize_log_entry(self, entry): 脱敏处理移除或混淆敏感信息 sanitized entry.copy() # 示例隐藏可能的 API 密钥片段 if input_messages in sanitized: for msg in sanitized[input_messages]: if content in msg: msg[content] re.sub(rsk-\w{20}, sk-***, msg[content]) return sanitized # 使用示例 logger LLMInteractionLogger() def monitored_llm_call(messages, model, client): start_time time.time() try: response client.chat.completions.create( modelmodel, messagesmessages ) duration time.time() - start_time logger.log_interaction(messages, response, model, duration, successTrue) return response except Exception as e: duration time.time() - start_time logger.log_interaction(messages, e, model, duration, successFalse) raise6. 常见问题排查与优化建议在实际项目中LLM 输入环节经常会遇到各种问题。下面是一些典型场景的排查路径。6.1 输入相关错误排查表问题现象可能原因检查方式处理建议模型返回无关内容系统提示词未生效或被覆盖检查消息列表中 system 角色消息的位置和内容确保 system 消息是第一条内容清晰明确对话上下文丢失超出上下文窗口限制计算消息总令牌数检查模型限制实现对话历史管理移除最早消息或进行摘要模型拒绝回答输入格式错误或包含敏感词检查消息结构验证内容规范性确保消息角色正确清理输入内容处理长文档时性能差单次输入过长或分块策略不合理监控令牌使用量分析分块大小优化文本分块策略优先输入相关段落多模态输入失败图像格式不支持或编码错误验证图像格式、大小和编码方式转换为模型支持的格式检查 base64 编码6.2 输入管道性能优化# 异步处理提高吞吐量 import asyncio from concurrent.futures import ThreadPoolExecutor class AsyncInputProcessor: def __init__(self, max_workers5): self.executor ThreadPoolExecutor(max_workersmax_workers) async def process_input_batch(self, inputs): 批量处理输入如图片转 base64、文本分块等 loop asyncio.get_event_loop() # 将同步的 CPU 密集型任务放到线程池中执行 tasks [ loop.run_in_executor(self.executor, self.process_single_input, input_data) for input_data in inputs ] results await asyncio.gather(*tasks, return_exceptionsTrue) return results def process_single_input(self, input_data): 处理单个输入同步方法 # 这里可以是文本清理、文件解析等操作 if isinstance(input_data, str): return validate_and_clean_input(input_data) else: # 处理文件或其他类型输入 return input_data # 使用示例 async def main(): processor AsyncInputProcessor() inputs [输入1, 输入2, 输入3] results await processor.process_input_batch(inputs) print(f处理完成 {len(results)} 个输入) # asyncio.run(main())6.3 安全最佳实践输入验证层层把关前端验证防止明显恶意输入后端验证严格的内容检查和长度限制模型前验证最终的内容清理和格式化敏感信息过滤def filter_sensitive_info(text): 过滤可能的敏感信息 patterns [ r\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b, # 信用卡号 r\b\d{3}[- ]?\d{2}[- ]?\d{4}\b, # SSN rsk-\w{20,}, # API 密钥 ] for pattern in patterns: text re.sub(pattern, [FILTERED], text) return text速率限制与滥用防护实现用户级别的请求频率限制监控异常输入模式设置合理的超时时间构建可靠的 LLM 输入管道需要综合考虑格式规范、内容处理、上下文管理和生产环境要求。从准确的文件解析到智能的对话管理从多模态支持到健壮的错误处理每个环节都直接影响最终的用户体验和系统稳定性。在实际项目中建议从最小可行方案开始逐步迭代加入更高级的功能和保障机制。
