1. 项目概述为什么我们要“最小化”LangChain如果你最近在折腾大语言模型应用开发大概率听过LangChain这个名字。它像是一个功能齐全的“瑞士军刀”提供了从连接模型、管理记忆、调用工具到构建复杂Agent工作流的一整套工具链。对于快速原型验证和复杂应用构建LangChain无疑是强大的。但不知道你有没有过这种感觉当你只是想快速验证一个简单的想法比如让LLM根据你的本地文档回答几个问题或者做一个极简的聊天机器人时引入完整的LangChain框架总有种“杀鸡用牛刀”的沉重感。你需要安装一堆依赖理解它抽象出来的各种概念Chains, Agents, Memory, Retrievers配置文件可能比你的核心逻辑代码还长。这就是我们动手“从零实现一个最小版LangChain”的初衷。这个“最小版”的目标不是复刻LangChain的所有功能而是抓住其最核心的设计思想——模块化编排LLM调用、工具使用和外部数据——并用最精简的代码实现一个可运行、可理解、可扩展的骨架。通过这个过程你能真正吃透LLM应用框架底层在做什么而不是停留在调包的层面。当你的简单需求被一个轻量级框架满足时部署会更简单调试会更直观你对整个流程的控制力也会更强。无论你是刚接触LLM应用的开发者还是想深入理解Agent/RAG原理的实践者这个“造轮子”的过程都会让你受益匪浅。2. 核心架构设计拆解LangChain的“三驾马车”LangChain虽然庞大但其核心思想可以归结为三个关键组件的协同工作LLM大语言模型、Tools工具和Memory记忆并通过一个编排器Orchestrator来调度它们。我们的最小版将围绕这三点展开。2.1 LLM模块不仅仅是API调用器在LangChain中LLM模块是一个抽象层它统一了不同模型提供商如OpenAI、Anthropic、本地部署的模型的调用接口。我们最小版的核心任务之一就是实现这个抽象。首先我们定义一个基础的LLM类。它不应该关心你最终用的是GPT-4还是Claude而是提供一个统一的generate方法。from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional class BaseLLM(ABC): LLM抽象基类。所有具体的LLM实现如OpenAI、本地模型都应继承此类。 abstractmethod def generate(self, prompt: str, **kwargs) - str: 核心生成方法。 Args: prompt: 输入的提示词文本。 **kwargs: 模型特定的参数如temperature, max_tokens等。 Returns: 模型生成的文本。 pass abstractmethod def get_num_tokens(self, text: str) - int: 计算文本的token数量对于上下文窗口管理很重要。 pass接下来我们实现一个具体的OpenAI封装。这里的关键点在于我们要处理API密钥、模型名称、以及将通用的参数如temperature映射到OpenAI API的具体参数。import openai from typing import Any class OpenAILanguageModel(BaseLLM): OpenAI LLM的具体实现。 def __init__(self, model_name: str gpt-3.5-turbo, api_key: Optional[str] None, **default_kwargs): 初始化。 Args: model_name: 如 gpt-3.5-turbo, gpt-4。 api_key: OpenAI API密钥。如果为None会尝试从环境变量OPENAI_API_KEY读取。 **default_kwargs: 默认的调用参数如 temperature0.7。 self.model_name model_name self.default_kwargs default_kwargs # 简单的密钥管理生产环境建议使用更安全的方式 self.client openai.OpenAI(api_keyapi_key or openai.api_key) def generate(self, prompt: str, **kwargs) - str: # 合并默认参数和本次调用参数 call_kwargs {**self.default_kwargs, **kwargs} # 构造OpenAI API需要的消息格式对于Chat模型 messages [{role: user, content: prompt}] try: response self.client.chat.completions.create( modelself.model_name, messagesmessages, **call_kwargs ) return response.choices[0].message.content except openai.APIError as e: # 简单的错误处理实际项目中需要更完善 return fLLM API调用错误: {e} def get_num_tokens(self, text: str) - int: # 这是一个简化版本。OpenAI提供了tiktoken库进行精确计算。 # 这里为了最小化我们用一个粗略的估算英文大约1个token对应4个字符。 # 强烈建议在实际项目中使用 tiktoken。 return len(text) // 4注意这里的get_num_tokens是极简的估算仅用于演示。在实际的RAG或长上下文应用中token的精确计数至关重要直接关系到是否会超出模型上下文窗口以及API成本。对于OpenAI模型务必使用tiktoken库。我们在这里简化是为了保持项目核心的清晰。通过这样的设计如果我们明天想接入Anthropic的Claude只需要再创建一个ClaudeLLM类继承BaseLLM并实现其generate方法即可。应用的其他部分如后面的Agent完全不需要改动。这就是抽象的力量。2.2 工具模块让LLM拥有“手和脚”LLM本身是“大脑”它擅长思考和生成文本但不擅长执行具体动作比如搜索网络、查询数据库、运行代码。Tools就是LLM的“手和脚”。LangChain的Tools设计非常巧妙每个工具都有一个清晰的描述LLM根据描述决定在何时、使用哪个工具。我们先定义工具的基类。一个工具最核心的就是它的name、description和_run方法。from abc import ABC, abstractmethod from typing import Any, Dict class BaseTool(ABC): 工具抽象基类。 name: str description: str abstractmethod def _run(self, input_text: str) - str: 工具的执行逻辑。 Args: input_text: LLM解析后决定传入给工具的参数。 Returns: 工具执行的结果通常是一个字符串。 pass def run(self, input_text: str) - str: 对_run的包装可以在这里添加日志、错误处理等通用逻辑。 print(f[工具调用] {self.name}: {input_text}) try: result self._run(input_text) print(f[工具结果] {self.name}: {result[:100]}...) # 打印前100字符 return result except Exception as e: error_msg f工具 {self.name} 执行失败: {e} print(f[工具错误] {error_msg}) return error_msg现在让我们实现两个最常见的工具一个网络搜索工具模拟和一个计算器工具。import json import math import requests from typing import Any class CalculatorTool(BaseTool): 一个简单的计算器工具能执行基础数学运算。 name calculator description 用于执行数学计算。输入应该是一个数学表达式例如 2 3 * 4 或 sqrt(16)。 def _run(self, input_text: str) - str: # 警告使用eval有安全风险这里仅用于演示。 # 在生产环境中必须使用安全的表达式求值库如 ast.literal_eval 配合自定义操作符解析。 try: # 为表达式添加一些安全的数学函数 allowed_names {k: v for k, v in math.__dict__.items() if not k.startswith(_)} # 非常简陋的“安全”检查实际不可靠 result eval(input_text, {__builtins__: {}}, allowed_names) return str(result) except Exception as e: return f计算错误: {e} class WebSearchTool(BaseTool): 模拟的网络搜索工具。在实际项目中你会接入SerpAPI、Google Search API等。 name web_search description 用于搜索互联网上的最新信息。输入应该是一个搜索查询词。 def _run(self, query: str) - str: # 这里是模拟返回。真实情况需要调用搜索API。 # 例如你可以用 requests 调用 SerpAPI 或 DuckDuckGo Instant Answer API。 mock_results { 今天北京的天气: 北京2023年10月27日晴气温5-18°C西北风3-4级。, Python的最新版本: 截至2023年10月Python的最新稳定版本是3.12.0。, 未知查询: f已收到搜索请求: {query}。此为模拟工具返回固定信息。 } # 简单匹配实际应用中应使用更复杂的逻辑或真实API for key, value in mock_results.items(): if key in query: return value return mock_results[未知查询]实操心得工具描述description是Agent能正确使用工具的关键。描述必须清晰、准确说明工具的功能和期望的输入格式。例如calculator的描述明确指出“输入应该是一个数学表达式”。LLM会根据这个描述来格式化它的输出。写得模糊的描述会导致LLM错误调用工具。2.3 记忆模块会话的上下文基石没有记忆的LLM对话就像金鱼只有7秒。Memory模块负责保存和检索对话历史。最简单的记忆是ConversationBufferMemory它就像一个不断增长的列表保存所有过往的对话。from typing import List, Dict class ConversationBufferMemory: 简单的对话缓冲记忆保存完整的对话历史。 def __init__(self): self.buffer: List[Dict[str, str]] [] # 每个元素是 {role: user/assistant, content: ...} def add_user_message(self, message: str): 添加用户消息到记忆。 self.buffer.append({role: user, content: message}) def add_ai_message(self, message: str): 添加AI助手消息到记忆。 self.buffer.append({role: assistant, content: message}) def get_history_as_text(self, max_tokens: Optional[int] None) - str: 将历史记录格式化为文本用于构造LLM的提示词。 Args: max_tokens: 如果提供会尝试截断历史以不超过大致token数。 Returns: 格式化后的历史字符串。 history_text for turn in self.buffer: role Human if turn[role] user else Assistant history_text f{role}: {turn[content]}\n # 极简的token截断逻辑生产环境需用tiktoken精确计算并优先丢弃最早的历史 if max_tokens: estimated_tokens len(history_text) // 4 if estimated_tokens max_tokens: # 简单粗暴地按字符截断这并不科学 chars_to_keep max_tokens * 4 history_text ... history_text[-chars_to_keep:] return history_text def clear(self): 清空记忆。 self.buffer.clear()这个记忆模块非常基础。在真正的LangChain或复杂应用中你可能会用到ConversationSummaryMemory不是保存所有对话而是让LLM定期总结历史节省token。VectorStoreRetrieverMemory将历史对话存入向量数据库根据当前问题语义检索相关历史实现“长期记忆”。Window Buffer Memory只保留最近N轮对话。对于我们最小版的目标缓冲记忆已经足够演示核心概念记忆是提示词的一部分。在每次调用LLM前我们会把历史记录拼接到当前的用户问题前形成完整的上下文。3. 智能体引擎让LLM学会调用工具有了LLM、工具和记忆现在需要最核心的“大脑”——智能体Agent。它的职责是理解用户请求决定是否需要使用工具如果需要选择正确的工具并生成调用参数解析工具返回的结果并最终组织语言回复给用户。3.1 智能体的决策循环ReAct模式我们将实现一个基于ReAct (Reasoning Acting)模式的智能体。这是当前最流行且有效的Agent范式之一。其核心思想是让LLM以“思考-行动-观察”的循环来工作。思考ThinkLLM分析当前情况用户问题、对话历史、可用工具决定下一步该做什么直接回答还是调用某个工具。行动Act如果决定调用工具LLM必须以严格的格式输出例如Action: 工具名\nAction Input: 工具输入。观察Observe执行工具获得结果。将结果Observation: ...反馈给LLM。循环LLM根据观察结果再次进行“思考”直到它认为可以给出最终答案然后输出Final Answer: ...。我们需要设计一个提示词Prompt来让LLM学会这个循环。这个提示词需要包含对AI助手角色的定义。可用工具的名称和描述。对话历史。当前的用户输入。严格的输出格式要求。def _build_agent_prompt(tools: List[BaseTool], history: str, human_input: str) - str: 构造驱动Agent的提示词。 tools_description \n.join([f- {tool.name}: {tool.description} for tool in tools]) prompt f你是一个有帮助的AI助手可以访问以下工具来帮助你完成任务 {tools_description} 如果你需要调用工具请严格按照以下格式输出 Action: 工具名 Action Input: 工具的输入 人们会看到你的思考过程所以在你给出最终答案前请始终使用上述格式来调用工具。 工具调用返回的结果会以“Observation:”为前缀提供给你。 开始 {history} Human: {human_input} Assistant: return prompt3.2 解析LLM输出与执行调度LLM根据提示词生成了文本我们需要从中解析出它是想调用工具还是直接回答。import re class SimpleReActAgent: 一个极简的ReAct模式智能体。 def __init__(self, llm: BaseLLM, tools: List[BaseTool], memory: ConversationBufferMemory): self.llm llm self.tools {tool.name: tool for tool in tools} # 转为字典便于查找 self.memory memory self.max_iterations 5 # 防止无限循环 def run(self, user_input: str) - str: 执行单轮对话。 print(f\n[用户输入] {user_input}) self.memory.add_user_message(user_input) history_text self.memory.get_history_as_text(max_tokens1500) # 限制历史长度 iteration 0 while iteration self.max_iterations: iteration 1 print(f\n[Agent循环] 第{iteration}次迭代) # 1. 构建提示词并调用LLM prompt _build_agent_prompt(list(self.tools.values()), history_text, user_input) llm_output self.llm.generate(prompt, temperature0, max_tokens500) print(f[LLM原始输出]\n{llm_output}) # 2. 解析LLM输出 action_match re.search(rAction:\s*(.), llm_output) action_input_match re.search(rAction Input:\s*(.), llm_output) final_answer_match re.search(rFinal Answer:\s*(.), llm_output, re.DOTALL) # re.DOTALL让.匹配换行 # 3. 情况A: LLM决定给出最终答案 if final_answer_match: final_answer final_answer_match.group(1).strip() self.memory.add_ai_message(final_answer) print(f[最终答案] {final_answer}) return final_answer # 4. 情况B: LLM决定调用工具 elif action_match and action_input_match: action action_match.group(1).strip() action_input action_input_match.group(1).strip() if action not in self.tools: error_msg f未知工具: {action}. 可用工具: {list(self.tools.keys())} observation error_msg else: # 执行工具 tool self.tools[action] observation tool.run(action_input) # 将“观察”结果添加到历史中用于下一轮循环 # 注意这里我们把“Action: ...”和“Observation: ...”都加到历史模拟LLM看到完整过程。 # 更精细的实现可能只加Observation。 history_text f\nAssistant: Action: {action}\nAction Input: {action_input}\nObservation: {observation}\n # 注意我们暂时不把这次中间过程存入memory.buffer只用于本次循环的上下文。 # 最终答案才会被存入memory。 print(f[工具观察] {observation[:200]}) # 5. 情况C: LLM输出不符合任何格式解析失败 else: error_msg LLM的输出格式无法解析既不是工具调用也不是最终答案。 observation error_msg history_text f\nAssistant: {llm_output}\nObservation: {observation}\n print(f[解析错误] {error_msg}) # 循环超过最大次数强制结束 final_answer 抱歉我尝试了多次但仍未能解决问题。请尝试重新表述您的问题。 self.memory.add_ai_message(final_answer) return final_answer这个SimpleReActAgent类就是我们的核心引擎。它在一个循环中用当前历史包含之前的工具调用和观察和用户问题构建提示词。调用LLM。尝试解析LLM的输出。如果是最终答案就存储并返回。如果是工具调用就执行工具并把结果Observation拼接到历史上下文中进入下一轮循环。注意事项这里有一个关键设计选择在while循环中我们不断用history_text累加Observation但只有最终答案才通过self.memory.add_ai_message存入长期记忆。这意味着工具调用的中间过程不会被永久记住。这通常是个好设计可以避免记忆被大量中间步骤污染。下次用户问新问题时Agent看到的历史只有之前的最终问答没有中间的工具调用细节。4. 检索增强生成实现让LLM“读懂”你的文档RAG是当前让LLM获取私有、最新知识的最主流方法。其核心流程是“检索-生成”先从你的文档库知识库中找出与问题相关的片段然后将这些片段和问题一起交给LLM让它基于这些上下文生成答案。4.1 文档加载与文本分割首先我们需要处理原始文档。文档可能来自PDF、Word、网页或纯文本。为了最小化我们假设文档已经是纯文本。第一步是将长文本分割成更小的“块”Chunks以便后续嵌入和检索。from typing import List import re class TextSplitter: 简单的文本分割器。 def __init__(self, chunk_size: int 500, chunk_overlap: int 50): Args: chunk_size: 每个文本块的最大字符数。 chunk_overlap: 相邻块之间的重叠字符数用于保持上下文连贯。 self.chunk_size chunk_size self.chunk_overlap chunk_overlap def split_text(self, text: str) - List[str]: 将文本分割成块。 # 这是一个基于字符的简单分割。更高级的分割器会考虑句子、段落甚至语义边界。 if len(text) self.chunk_size: return [text] chunks [] start 0 while start len(text): end start self.chunk_size # 如果还没到文本末尾尝试在空格或标点处截断避免切断单词。 if end len(text): # 查找最后一个空格或句号 while end start and text[end] not in ( , ., 。, !, ?, \n): end - 1 # 如果没找到合适的断点就强制在chunk_size处截断 if end start: end start self.chunk_size else: end len(text) chunk text[start:end] chunks.append(chunk) # 移动起始位置考虑重叠 start end - self.chunk_overlap return chunks4.2 向量化与存储构建知识库分割后的文本块需要转换成向量嵌入并存储到向量数据库中以便进行相似度搜索。我们使用sentence-transformers库来生成嵌入并用一个简单的内存字典模拟向量数据库。import numpy as np from typing import List, Tuple, Dict # 注意需要先安装 sentence-transformers: pip install sentence-transformers from sentence_transformers import SentenceTransformer class SimpleVectorStore: 一个极简的、基于内存的向量存储。 def __init__(self, embedding_model_name: str all-MiniLM-L6-v2): Args: embedding_model_name: 用于生成文本嵌入的模型名称。 # 加载嵌入模型首次使用会下载 self.embedder SentenceTransformer(embedding_model_name) self.documents: List[str] [] # 存储原始文本 self.embeddings: np.ndarray None # 存储对应的向量 def add_documents(self, texts: List[str]): 将文本列表添加到向量库中。 print(f正在为 {len(texts)} 个文本块生成嵌入...) new_embeddings self.embedder.encode(texts, convert_to_numpyTrue) if self.embeddings is None: self.embeddings new_embeddings else: self.embeddings np.vstack([self.embeddings, new_embeddings]) self.documents.extend(texts) print(f向量库更新完成现有 {len(self.documents)} 个文档。) def similarity_search(self, query: str, k: int 3) - List[Tuple[str, float]]: 根据查询文本返回最相似的k个文档及其相似度分数。 Args: query: 查询文本。 k: 返回的最相似文档数量。 Returns: 列表元素为(文档文本, 相似度分数)。 if len(self.documents) 0: return [] # 将查询文本转换为向量 query_embedding self.embedder.encode([query], convert_to_numpyTrue)[0] # 计算余弦相似度 (向量点积 / (模长乘积))因为我们的嵌入是归一化的所以点积即余弦相似度 # 这里假设 self.embeddings 是归一化的sentence-transformers 默认输出是归一化的。 similarities np.dot(self.embeddings, query_embedding) # 获取相似度最高的k个索引 top_k_indices np.argsort(similarities)[-k:][::-1] # 从高到低排序 results [] for idx in top_k_indices: results.append((self.documents[idx], float(similarities[idx]))) return results4.3 RAG链路的组装现在我们将向量检索和LLM生成组合成一个完整的RAG流程。class SimpleRAGChain: 一个简单的RAG链集成了检索和生成。 def __init__(self, vector_store: SimpleVectorStore, llm: BaseLLM): self.vector_store vector_store self.llm llm def query(self, question: str, k: int 3) - str: 针对问题检索相关文档并生成答案。 Args: question: 用户问题。 k: 检索的文档数量。 Returns: LLM生成的答案。 # 1. 检索 print(f[RAG] 正在检索与问题相关的文档...) retrieved_docs self.vector_store.similarity_search(question, kk) if not retrieved_docs: context 未找到相关文档。 else: # 将检索到的文档拼接成上下文 context_parts [] for i, (doc, score) in enumerate(retrieved_docs): context_parts.append(f[文档片段 {i1}, 相关度: {score:.3f}]\n{doc}) context \n\n.join(context_parts) print(f[RAG] 检索到的上下文前500字符:\n{context[:500]}...\n) # 2. 构建提示词 prompt f请基于以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题请直接说明你不知道不要编造信息。 上下文信息 {context} 问题{question} 请根据上下文给出答案 # 3. 生成 print(f[RAG] 正在生成答案...) answer self.llm.generate(prompt, temperature0.1) # 低temperature使答案更基于上下文 return answer5. 实战演练组装你的第一个AI助手现在让我们把上面所有的模块像乐高一样搭起来创建一个具备工具调用和文档问答能力的AI助手。5.1 环境准备与模块导入首先确保你安装了必要的库。创建一个新的Python虚拟环境是个好习惯。# 建议在虚拟环境中操作 pip install openai sentence-transformers numpy requests然后将我们之前编写的所有类BaseLLM,OpenAILanguageModel,BaseTool,CalculatorTool,WebSearchTool,ConversationBufferMemory,SimpleReActAgent,TextSplitter,SimpleVectorStore,SimpleRAGChain放在一个Python文件中或者导入它们。5.2 场景一创建一个会使用工具的对话Agentdef demo_agent(): 演示一个能使用计算器和搜索工具的智能体。 print( 启动工具调用智能体演示 ) # 1. 初始化组件 llm OpenAILanguageModel(model_namegpt-3.5-turbo, temperature0) # 使用gpt-3.5-turbo确定性高 calculator CalculatorTool() searcher WebSearchTool() memory ConversationBufferMemory() # 2. 创建智能体 agent SimpleReActAgent(llmllm, tools[calculator, searcher], memorymemory) # 3. 进行多轮对话 queries [ 2的10次方是多少, 今天北京的天气怎么样, 那么上海呢, # 这个测试记忆是否有效 请计算一下如果北京气温是18度换算成华氏度是多少 # 这个需要结合搜索天气和计算 ] for query in queries: print(f\n{*40}) answer agent.run(query) print(f\n[助手回复] {answer}) print(f{*40}) if __name__ __main__: # 请确保已设置环境变量 OPENAI_API_KEY demo_agent()运行这段代码你会看到控制台打印出Agent的思考过程它如何解析问题、选择工具、执行工具、并根据结果进行下一步。例如对于最后一个复合问题它可能会先调用web_search获取“北京气温18度”然后调用calculator进行(18 * 9/5) 32的计算。5.3 场景二构建一个私人知识库问答系统假设你有一份公司产品手册的文本文件handbook.txt现在想基于它来回答问题。def demo_rag(): 演示基于私人文档的RAG问答。 print(\n 启动RAG知识库演示 ) # 1. 准备LLM和向量库 llm OpenAILanguageModel(model_namegpt-3.5-turbo) vector_store SimpleVectorStore() # 2. 加载并处理文档 with open(handbook.txt, r, encodingutf-8) as f: full_text f.read() splitter TextSplitter(chunk_size300, chunk_overlap50) chunks splitter.split_text(full_text) print(f文档被分割成 {len(chunks)} 个块。) # 3. 构建向量知识库 vector_store.add_documents(chunks) # 4. 创建RAG链 rag_chain SimpleRAGChain(vector_storevector_store, llmllm) # 5. 进行问答 questions [ 我们公司的主要产品是什么, 产品的保修期是多久, 如何联系客服 ] for q in questions: print(f\n{*40}) print(f[问题] {q}) answer rag_chain.query(q, k2) # 每次检索2个最相关的片段 print(f[RAG答案] {answer}) print(f{*40}) if __name__ __main__: demo_rag()这个演示展示了RAG的核心价值LLM的答案完全来源于你提供的handbook.txt避免了“幻觉”胡编乱造并且能处理模型训练数据中不存在的最新或私有信息。6. 避坑指南与进阶思考在实现和运行这个最小版框架的过程中你肯定会遇到各种问题。这里分享一些我踩过的坑和进阶思路。6.1 常见问题与排查LLM不按格式输出这是开发Agent时最常见的问题。我们的解析器依赖正则表达式匹配Action:和Final Answer:。原因提示词不够清晰或者LLM的temperature参数太高导致输出随机。解决将temperature设为0确保输出的确定性。在提示词中更加强调格式甚至提供几个清晰的示例Few-shot Prompting。使用更强大的模型如GPT-4通常格式遵循能力更好。实现一个“后处理”或“重试”逻辑如果解析失败可以提示LLM“请严格按照指定格式重新输出”。工具调用陷入死循环Agent可能反复调用同一个工具或者在不同工具间来回切换无法得出最终答案。原因工具结果未能满足LLM的“思考”或者LLM陷入了逻辑循环。解决设置max_iterations如我们代码中的5次来强制终止。优化工具的描述使其功能边界更清晰。在提示词中鼓励LLM在获得足够信息后及时给出Final Answer。RAG检索结果不相关LLM基于不相关的上下文生成了错误答案。原因文本分割不合理或者嵌入模型/检索方式不适合你的领域。解决优化分块不要简单按字符数分。尝试按段落、按标题分或者使用更高级的语义分割器。尝试不同嵌入模型all-MiniLM-L6-v2是通用模型。对于特定领域如医学、法律使用在该领域微调过的嵌入模型效果更好。引入重排序Re-ranking先用向量检索出Top 20个候选再用一个更精细的交叉编码器模型对它们进行重排序选出Top 3给LLM准确性会大幅提升。优化查询对用户原始问题进行改写或扩展Query Expansion再用于检索。Token超限错误当对话历史或检索的上下文太长时会超过LLM的上下文窗口。解决为记忆模块实现更智能的截断或总结策略如ConversationSummaryMemory。在RAG中限制检索返回的文本块数量k和每个块的大小chunk_size。精确计算token使用tiktoken等库并在构造提示词前主动截断。6.2 从“最小版”到“可用版”的进阶路径我们这个最小版实现了核心思想但要用于生产还需要在很多方面加固和扩展健壮性错误处理为LLM API调用、工具执行添加更完善的异常捕获和重试机制。解析鲁棒性用更灵活的解析器如基于语法替代简单的正则匹配提高对LLM输出格式波动的容忍度。超时控制为LLM调用和工具执行设置超时防止卡死。功能扩展更多工具实现文件读写、数据库查询、代码执行等工具。多模态集成视觉模型让Agent能“看”图片并描述或分析。复杂工作流引入类似LangGraph的图编排支持循环、分支、并行等复杂Agent逻辑。记忆升级实现向量记忆、摘要记忆让Agent拥有更强大的长期记忆能力。工程化配置化将模型参数、工具列表、提示词模板等外置到配置文件如YAML。可观测性加入详细的日志记录追踪每一次LLM调用、工具执行和token消耗便于调试和成本分析。异步化使用asyncio将LLM调用和I/O密集型工具改为异步提升整体响应速度。通过这个从零构建的过程你应该对LangChain这类框架所解决的问题、其内部的核心抽象和运行机制有了更深刻的理解。下次当你再使用成熟的框架时你会更清楚每一行配置代码背后发生了什么也能在遇到问题时更快地定位到是提示词、工具、记忆还是Agent逻辑本身的问题。这才是“造轮子”最大的价值——不是为了替代而是为了透彻的理解。 从零实现最小化LangChain：深入理解LLM应用框架核心原理
1. 项目概述为什么我们要“最小化”LangChain如果你最近在折腾大语言模型应用开发大概率听过LangChain这个名字。它像是一个功能齐全的“瑞士军刀”提供了从连接模型、管理记忆、调用工具到构建复杂Agent工作流的一整套工具链。对于快速原型验证和复杂应用构建LangChain无疑是强大的。但不知道你有没有过这种感觉当你只是想快速验证一个简单的想法比如让LLM根据你的本地文档回答几个问题或者做一个极简的聊天机器人时引入完整的LangChain框架总有种“杀鸡用牛刀”的沉重感。你需要安装一堆依赖理解它抽象出来的各种概念Chains, Agents, Memory, Retrievers配置文件可能比你的核心逻辑代码还长。这就是我们动手“从零实现一个最小版LangChain”的初衷。这个“最小版”的目标不是复刻LangChain的所有功能而是抓住其最核心的设计思想——模块化编排LLM调用、工具使用和外部数据——并用最精简的代码实现一个可运行、可理解、可扩展的骨架。通过这个过程你能真正吃透LLM应用框架底层在做什么而不是停留在调包的层面。当你的简单需求被一个轻量级框架满足时部署会更简单调试会更直观你对整个流程的控制力也会更强。无论你是刚接触LLM应用的开发者还是想深入理解Agent/RAG原理的实践者这个“造轮子”的过程都会让你受益匪浅。2. 核心架构设计拆解LangChain的“三驾马车”LangChain虽然庞大但其核心思想可以归结为三个关键组件的协同工作LLM大语言模型、Tools工具和Memory记忆并通过一个编排器Orchestrator来调度它们。我们的最小版将围绕这三点展开。2.1 LLM模块不仅仅是API调用器在LangChain中LLM模块是一个抽象层它统一了不同模型提供商如OpenAI、Anthropic、本地部署的模型的调用接口。我们最小版的核心任务之一就是实现这个抽象。首先我们定义一个基础的LLM类。它不应该关心你最终用的是GPT-4还是Claude而是提供一个统一的generate方法。from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional class BaseLLM(ABC): LLM抽象基类。所有具体的LLM实现如OpenAI、本地模型都应继承此类。 abstractmethod def generate(self, prompt: str, **kwargs) - str: 核心生成方法。 Args: prompt: 输入的提示词文本。 **kwargs: 模型特定的参数如temperature, max_tokens等。 Returns: 模型生成的文本。 pass abstractmethod def get_num_tokens(self, text: str) - int: 计算文本的token数量对于上下文窗口管理很重要。 pass接下来我们实现一个具体的OpenAI封装。这里的关键点在于我们要处理API密钥、模型名称、以及将通用的参数如temperature映射到OpenAI API的具体参数。import openai from typing import Any class OpenAILanguageModel(BaseLLM): OpenAI LLM的具体实现。 def __init__(self, model_name: str gpt-3.5-turbo, api_key: Optional[str] None, **default_kwargs): 初始化。 Args: model_name: 如 gpt-3.5-turbo, gpt-4。 api_key: OpenAI API密钥。如果为None会尝试从环境变量OPENAI_API_KEY读取。 **default_kwargs: 默认的调用参数如 temperature0.7。 self.model_name model_name self.default_kwargs default_kwargs # 简单的密钥管理生产环境建议使用更安全的方式 self.client openai.OpenAI(api_keyapi_key or openai.api_key) def generate(self, prompt: str, **kwargs) - str: # 合并默认参数和本次调用参数 call_kwargs {**self.default_kwargs, **kwargs} # 构造OpenAI API需要的消息格式对于Chat模型 messages [{role: user, content: prompt}] try: response self.client.chat.completions.create( modelself.model_name, messagesmessages, **call_kwargs ) return response.choices[0].message.content except openai.APIError as e: # 简单的错误处理实际项目中需要更完善 return fLLM API调用错误: {e} def get_num_tokens(self, text: str) - int: # 这是一个简化版本。OpenAI提供了tiktoken库进行精确计算。 # 这里为了最小化我们用一个粗略的估算英文大约1个token对应4个字符。 # 强烈建议在实际项目中使用 tiktoken。 return len(text) // 4注意这里的get_num_tokens是极简的估算仅用于演示。在实际的RAG或长上下文应用中token的精确计数至关重要直接关系到是否会超出模型上下文窗口以及API成本。对于OpenAI模型务必使用tiktoken库。我们在这里简化是为了保持项目核心的清晰。通过这样的设计如果我们明天想接入Anthropic的Claude只需要再创建一个ClaudeLLM类继承BaseLLM并实现其generate方法即可。应用的其他部分如后面的Agent完全不需要改动。这就是抽象的力量。2.2 工具模块让LLM拥有“手和脚”LLM本身是“大脑”它擅长思考和生成文本但不擅长执行具体动作比如搜索网络、查询数据库、运行代码。Tools就是LLM的“手和脚”。LangChain的Tools设计非常巧妙每个工具都有一个清晰的描述LLM根据描述决定在何时、使用哪个工具。我们先定义工具的基类。一个工具最核心的就是它的name、description和_run方法。from abc import ABC, abstractmethod from typing import Any, Dict class BaseTool(ABC): 工具抽象基类。 name: str description: str abstractmethod def _run(self, input_text: str) - str: 工具的执行逻辑。 Args: input_text: LLM解析后决定传入给工具的参数。 Returns: 工具执行的结果通常是一个字符串。 pass def run(self, input_text: str) - str: 对_run的包装可以在这里添加日志、错误处理等通用逻辑。 print(f[工具调用] {self.name}: {input_text}) try: result self._run(input_text) print(f[工具结果] {self.name}: {result[:100]}...) # 打印前100字符 return result except Exception as e: error_msg f工具 {self.name} 执行失败: {e} print(f[工具错误] {error_msg}) return error_msg现在让我们实现两个最常见的工具一个网络搜索工具模拟和一个计算器工具。import json import math import requests from typing import Any class CalculatorTool(BaseTool): 一个简单的计算器工具能执行基础数学运算。 name calculator description 用于执行数学计算。输入应该是一个数学表达式例如 2 3 * 4 或 sqrt(16)。 def _run(self, input_text: str) - str: # 警告使用eval有安全风险这里仅用于演示。 # 在生产环境中必须使用安全的表达式求值库如 ast.literal_eval 配合自定义操作符解析。 try: # 为表达式添加一些安全的数学函数 allowed_names {k: v for k, v in math.__dict__.items() if not k.startswith(_)} # 非常简陋的“安全”检查实际不可靠 result eval(input_text, {__builtins__: {}}, allowed_names) return str(result) except Exception as e: return f计算错误: {e} class WebSearchTool(BaseTool): 模拟的网络搜索工具。在实际项目中你会接入SerpAPI、Google Search API等。 name web_search description 用于搜索互联网上的最新信息。输入应该是一个搜索查询词。 def _run(self, query: str) - str: # 这里是模拟返回。真实情况需要调用搜索API。 # 例如你可以用 requests 调用 SerpAPI 或 DuckDuckGo Instant Answer API。 mock_results { 今天北京的天气: 北京2023年10月27日晴气温5-18°C西北风3-4级。, Python的最新版本: 截至2023年10月Python的最新稳定版本是3.12.0。, 未知查询: f已收到搜索请求: {query}。此为模拟工具返回固定信息。 } # 简单匹配实际应用中应使用更复杂的逻辑或真实API for key, value in mock_results.items(): if key in query: return value return mock_results[未知查询]实操心得工具描述description是Agent能正确使用工具的关键。描述必须清晰、准确说明工具的功能和期望的输入格式。例如calculator的描述明确指出“输入应该是一个数学表达式”。LLM会根据这个描述来格式化它的输出。写得模糊的描述会导致LLM错误调用工具。2.3 记忆模块会话的上下文基石没有记忆的LLM对话就像金鱼只有7秒。Memory模块负责保存和检索对话历史。最简单的记忆是ConversationBufferMemory它就像一个不断增长的列表保存所有过往的对话。from typing import List, Dict class ConversationBufferMemory: 简单的对话缓冲记忆保存完整的对话历史。 def __init__(self): self.buffer: List[Dict[str, str]] [] # 每个元素是 {role: user/assistant, content: ...} def add_user_message(self, message: str): 添加用户消息到记忆。 self.buffer.append({role: user, content: message}) def add_ai_message(self, message: str): 添加AI助手消息到记忆。 self.buffer.append({role: assistant, content: message}) def get_history_as_text(self, max_tokens: Optional[int] None) - str: 将历史记录格式化为文本用于构造LLM的提示词。 Args: max_tokens: 如果提供会尝试截断历史以不超过大致token数。 Returns: 格式化后的历史字符串。 history_text for turn in self.buffer: role Human if turn[role] user else Assistant history_text f{role}: {turn[content]}\n # 极简的token截断逻辑生产环境需用tiktoken精确计算并优先丢弃最早的历史 if max_tokens: estimated_tokens len(history_text) // 4 if estimated_tokens max_tokens: # 简单粗暴地按字符截断这并不科学 chars_to_keep max_tokens * 4 history_text ... history_text[-chars_to_keep:] return history_text def clear(self): 清空记忆。 self.buffer.clear()这个记忆模块非常基础。在真正的LangChain或复杂应用中你可能会用到ConversationSummaryMemory不是保存所有对话而是让LLM定期总结历史节省token。VectorStoreRetrieverMemory将历史对话存入向量数据库根据当前问题语义检索相关历史实现“长期记忆”。Window Buffer Memory只保留最近N轮对话。对于我们最小版的目标缓冲记忆已经足够演示核心概念记忆是提示词的一部分。在每次调用LLM前我们会把历史记录拼接到当前的用户问题前形成完整的上下文。3. 智能体引擎让LLM学会调用工具有了LLM、工具和记忆现在需要最核心的“大脑”——智能体Agent。它的职责是理解用户请求决定是否需要使用工具如果需要选择正确的工具并生成调用参数解析工具返回的结果并最终组织语言回复给用户。3.1 智能体的决策循环ReAct模式我们将实现一个基于ReAct (Reasoning Acting)模式的智能体。这是当前最流行且有效的Agent范式之一。其核心思想是让LLM以“思考-行动-观察”的循环来工作。思考ThinkLLM分析当前情况用户问题、对话历史、可用工具决定下一步该做什么直接回答还是调用某个工具。行动Act如果决定调用工具LLM必须以严格的格式输出例如Action: 工具名\nAction Input: 工具输入。观察Observe执行工具获得结果。将结果Observation: ...反馈给LLM。循环LLM根据观察结果再次进行“思考”直到它认为可以给出最终答案然后输出Final Answer: ...。我们需要设计一个提示词Prompt来让LLM学会这个循环。这个提示词需要包含对AI助手角色的定义。可用工具的名称和描述。对话历史。当前的用户输入。严格的输出格式要求。def _build_agent_prompt(tools: List[BaseTool], history: str, human_input: str) - str: 构造驱动Agent的提示词。 tools_description \n.join([f- {tool.name}: {tool.description} for tool in tools]) prompt f你是一个有帮助的AI助手可以访问以下工具来帮助你完成任务 {tools_description} 如果你需要调用工具请严格按照以下格式输出 Action: 工具名 Action Input: 工具的输入 人们会看到你的思考过程所以在你给出最终答案前请始终使用上述格式来调用工具。 工具调用返回的结果会以“Observation:”为前缀提供给你。 开始 {history} Human: {human_input} Assistant: return prompt3.2 解析LLM输出与执行调度LLM根据提示词生成了文本我们需要从中解析出它是想调用工具还是直接回答。import re class SimpleReActAgent: 一个极简的ReAct模式智能体。 def __init__(self, llm: BaseLLM, tools: List[BaseTool], memory: ConversationBufferMemory): self.llm llm self.tools {tool.name: tool for tool in tools} # 转为字典便于查找 self.memory memory self.max_iterations 5 # 防止无限循环 def run(self, user_input: str) - str: 执行单轮对话。 print(f\n[用户输入] {user_input}) self.memory.add_user_message(user_input) history_text self.memory.get_history_as_text(max_tokens1500) # 限制历史长度 iteration 0 while iteration self.max_iterations: iteration 1 print(f\n[Agent循环] 第{iteration}次迭代) # 1. 构建提示词并调用LLM prompt _build_agent_prompt(list(self.tools.values()), history_text, user_input) llm_output self.llm.generate(prompt, temperature0, max_tokens500) print(f[LLM原始输出]\n{llm_output}) # 2. 解析LLM输出 action_match re.search(rAction:\s*(.), llm_output) action_input_match re.search(rAction Input:\s*(.), llm_output) final_answer_match re.search(rFinal Answer:\s*(.), llm_output, re.DOTALL) # re.DOTALL让.匹配换行 # 3. 情况A: LLM决定给出最终答案 if final_answer_match: final_answer final_answer_match.group(1).strip() self.memory.add_ai_message(final_answer) print(f[最终答案] {final_answer}) return final_answer # 4. 情况B: LLM决定调用工具 elif action_match and action_input_match: action action_match.group(1).strip() action_input action_input_match.group(1).strip() if action not in self.tools: error_msg f未知工具: {action}. 可用工具: {list(self.tools.keys())} observation error_msg else: # 执行工具 tool self.tools[action] observation tool.run(action_input) # 将“观察”结果添加到历史中用于下一轮循环 # 注意这里我们把“Action: ...”和“Observation: ...”都加到历史模拟LLM看到完整过程。 # 更精细的实现可能只加Observation。 history_text f\nAssistant: Action: {action}\nAction Input: {action_input}\nObservation: {observation}\n # 注意我们暂时不把这次中间过程存入memory.buffer只用于本次循环的上下文。 # 最终答案才会被存入memory。 print(f[工具观察] {observation[:200]}) # 5. 情况C: LLM输出不符合任何格式解析失败 else: error_msg LLM的输出格式无法解析既不是工具调用也不是最终答案。 observation error_msg history_text f\nAssistant: {llm_output}\nObservation: {observation}\n print(f[解析错误] {error_msg}) # 循环超过最大次数强制结束 final_answer 抱歉我尝试了多次但仍未能解决问题。请尝试重新表述您的问题。 self.memory.add_ai_message(final_answer) return final_answer这个SimpleReActAgent类就是我们的核心引擎。它在一个循环中用当前历史包含之前的工具调用和观察和用户问题构建提示词。调用LLM。尝试解析LLM的输出。如果是最终答案就存储并返回。如果是工具调用就执行工具并把结果Observation拼接到历史上下文中进入下一轮循环。注意事项这里有一个关键设计选择在while循环中我们不断用history_text累加Observation但只有最终答案才通过self.memory.add_ai_message存入长期记忆。这意味着工具调用的中间过程不会被永久记住。这通常是个好设计可以避免记忆被大量中间步骤污染。下次用户问新问题时Agent看到的历史只有之前的最终问答没有中间的工具调用细节。4. 检索增强生成实现让LLM“读懂”你的文档RAG是当前让LLM获取私有、最新知识的最主流方法。其核心流程是“检索-生成”先从你的文档库知识库中找出与问题相关的片段然后将这些片段和问题一起交给LLM让它基于这些上下文生成答案。4.1 文档加载与文本分割首先我们需要处理原始文档。文档可能来自PDF、Word、网页或纯文本。为了最小化我们假设文档已经是纯文本。第一步是将长文本分割成更小的“块”Chunks以便后续嵌入和检索。from typing import List import re class TextSplitter: 简单的文本分割器。 def __init__(self, chunk_size: int 500, chunk_overlap: int 50): Args: chunk_size: 每个文本块的最大字符数。 chunk_overlap: 相邻块之间的重叠字符数用于保持上下文连贯。 self.chunk_size chunk_size self.chunk_overlap chunk_overlap def split_text(self, text: str) - List[str]: 将文本分割成块。 # 这是一个基于字符的简单分割。更高级的分割器会考虑句子、段落甚至语义边界。 if len(text) self.chunk_size: return [text] chunks [] start 0 while start len(text): end start self.chunk_size # 如果还没到文本末尾尝试在空格或标点处截断避免切断单词。 if end len(text): # 查找最后一个空格或句号 while end start and text[end] not in ( , ., 。, !, ?, \n): end - 1 # 如果没找到合适的断点就强制在chunk_size处截断 if end start: end start self.chunk_size else: end len(text) chunk text[start:end] chunks.append(chunk) # 移动起始位置考虑重叠 start end - self.chunk_overlap return chunks4.2 向量化与存储构建知识库分割后的文本块需要转换成向量嵌入并存储到向量数据库中以便进行相似度搜索。我们使用sentence-transformers库来生成嵌入并用一个简单的内存字典模拟向量数据库。import numpy as np from typing import List, Tuple, Dict # 注意需要先安装 sentence-transformers: pip install sentence-transformers from sentence_transformers import SentenceTransformer class SimpleVectorStore: 一个极简的、基于内存的向量存储。 def __init__(self, embedding_model_name: str all-MiniLM-L6-v2): Args: embedding_model_name: 用于生成文本嵌入的模型名称。 # 加载嵌入模型首次使用会下载 self.embedder SentenceTransformer(embedding_model_name) self.documents: List[str] [] # 存储原始文本 self.embeddings: np.ndarray None # 存储对应的向量 def add_documents(self, texts: List[str]): 将文本列表添加到向量库中。 print(f正在为 {len(texts)} 个文本块生成嵌入...) new_embeddings self.embedder.encode(texts, convert_to_numpyTrue) if self.embeddings is None: self.embeddings new_embeddings else: self.embeddings np.vstack([self.embeddings, new_embeddings]) self.documents.extend(texts) print(f向量库更新完成现有 {len(self.documents)} 个文档。) def similarity_search(self, query: str, k: int 3) - List[Tuple[str, float]]: 根据查询文本返回最相似的k个文档及其相似度分数。 Args: query: 查询文本。 k: 返回的最相似文档数量。 Returns: 列表元素为(文档文本, 相似度分数)。 if len(self.documents) 0: return [] # 将查询文本转换为向量 query_embedding self.embedder.encode([query], convert_to_numpyTrue)[0] # 计算余弦相似度 (向量点积 / (模长乘积))因为我们的嵌入是归一化的所以点积即余弦相似度 # 这里假设 self.embeddings 是归一化的sentence-transformers 默认输出是归一化的。 similarities np.dot(self.embeddings, query_embedding) # 获取相似度最高的k个索引 top_k_indices np.argsort(similarities)[-k:][::-1] # 从高到低排序 results [] for idx in top_k_indices: results.append((self.documents[idx], float(similarities[idx]))) return results4.3 RAG链路的组装现在我们将向量检索和LLM生成组合成一个完整的RAG流程。class SimpleRAGChain: 一个简单的RAG链集成了检索和生成。 def __init__(self, vector_store: SimpleVectorStore, llm: BaseLLM): self.vector_store vector_store self.llm llm def query(self, question: str, k: int 3) - str: 针对问题检索相关文档并生成答案。 Args: question: 用户问题。 k: 检索的文档数量。 Returns: LLM生成的答案。 # 1. 检索 print(f[RAG] 正在检索与问题相关的文档...) retrieved_docs self.vector_store.similarity_search(question, kk) if not retrieved_docs: context 未找到相关文档。 else: # 将检索到的文档拼接成上下文 context_parts [] for i, (doc, score) in enumerate(retrieved_docs): context_parts.append(f[文档片段 {i1}, 相关度: {score:.3f}]\n{doc}) context \n\n.join(context_parts) print(f[RAG] 检索到的上下文前500字符:\n{context[:500]}...\n) # 2. 构建提示词 prompt f请基于以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题请直接说明你不知道不要编造信息。 上下文信息 {context} 问题{question} 请根据上下文给出答案 # 3. 生成 print(f[RAG] 正在生成答案...) answer self.llm.generate(prompt, temperature0.1) # 低temperature使答案更基于上下文 return answer5. 实战演练组装你的第一个AI助手现在让我们把上面所有的模块像乐高一样搭起来创建一个具备工具调用和文档问答能力的AI助手。5.1 环境准备与模块导入首先确保你安装了必要的库。创建一个新的Python虚拟环境是个好习惯。# 建议在虚拟环境中操作 pip install openai sentence-transformers numpy requests然后将我们之前编写的所有类BaseLLM,OpenAILanguageModel,BaseTool,CalculatorTool,WebSearchTool,ConversationBufferMemory,SimpleReActAgent,TextSplitter,SimpleVectorStore,SimpleRAGChain放在一个Python文件中或者导入它们。5.2 场景一创建一个会使用工具的对话Agentdef demo_agent(): 演示一个能使用计算器和搜索工具的智能体。 print( 启动工具调用智能体演示 ) # 1. 初始化组件 llm OpenAILanguageModel(model_namegpt-3.5-turbo, temperature0) # 使用gpt-3.5-turbo确定性高 calculator CalculatorTool() searcher WebSearchTool() memory ConversationBufferMemory() # 2. 创建智能体 agent SimpleReActAgent(llmllm, tools[calculator, searcher], memorymemory) # 3. 进行多轮对话 queries [ 2的10次方是多少, 今天北京的天气怎么样, 那么上海呢, # 这个测试记忆是否有效 请计算一下如果北京气温是18度换算成华氏度是多少 # 这个需要结合搜索天气和计算 ] for query in queries: print(f\n{*40}) answer agent.run(query) print(f\n[助手回复] {answer}) print(f{*40}) if __name__ __main__: # 请确保已设置环境变量 OPENAI_API_KEY demo_agent()运行这段代码你会看到控制台打印出Agent的思考过程它如何解析问题、选择工具、执行工具、并根据结果进行下一步。例如对于最后一个复合问题它可能会先调用web_search获取“北京气温18度”然后调用calculator进行(18 * 9/5) 32的计算。5.3 场景二构建一个私人知识库问答系统假设你有一份公司产品手册的文本文件handbook.txt现在想基于它来回答问题。def demo_rag(): 演示基于私人文档的RAG问答。 print(\n 启动RAG知识库演示 ) # 1. 准备LLM和向量库 llm OpenAILanguageModel(model_namegpt-3.5-turbo) vector_store SimpleVectorStore() # 2. 加载并处理文档 with open(handbook.txt, r, encodingutf-8) as f: full_text f.read() splitter TextSplitter(chunk_size300, chunk_overlap50) chunks splitter.split_text(full_text) print(f文档被分割成 {len(chunks)} 个块。) # 3. 构建向量知识库 vector_store.add_documents(chunks) # 4. 创建RAG链 rag_chain SimpleRAGChain(vector_storevector_store, llmllm) # 5. 进行问答 questions [ 我们公司的主要产品是什么, 产品的保修期是多久, 如何联系客服 ] for q in questions: print(f\n{*40}) print(f[问题] {q}) answer rag_chain.query(q, k2) # 每次检索2个最相关的片段 print(f[RAG答案] {answer}) print(f{*40}) if __name__ __main__: demo_rag()这个演示展示了RAG的核心价值LLM的答案完全来源于你提供的handbook.txt避免了“幻觉”胡编乱造并且能处理模型训练数据中不存在的最新或私有信息。6. 避坑指南与进阶思考在实现和运行这个最小版框架的过程中你肯定会遇到各种问题。这里分享一些我踩过的坑和进阶思路。6.1 常见问题与排查LLM不按格式输出这是开发Agent时最常见的问题。我们的解析器依赖正则表达式匹配Action:和Final Answer:。原因提示词不够清晰或者LLM的temperature参数太高导致输出随机。解决将temperature设为0确保输出的确定性。在提示词中更加强调格式甚至提供几个清晰的示例Few-shot Prompting。使用更强大的模型如GPT-4通常格式遵循能力更好。实现一个“后处理”或“重试”逻辑如果解析失败可以提示LLM“请严格按照指定格式重新输出”。工具调用陷入死循环Agent可能反复调用同一个工具或者在不同工具间来回切换无法得出最终答案。原因工具结果未能满足LLM的“思考”或者LLM陷入了逻辑循环。解决设置max_iterations如我们代码中的5次来强制终止。优化工具的描述使其功能边界更清晰。在提示词中鼓励LLM在获得足够信息后及时给出Final Answer。RAG检索结果不相关LLM基于不相关的上下文生成了错误答案。原因文本分割不合理或者嵌入模型/检索方式不适合你的领域。解决优化分块不要简单按字符数分。尝试按段落、按标题分或者使用更高级的语义分割器。尝试不同嵌入模型all-MiniLM-L6-v2是通用模型。对于特定领域如医学、法律使用在该领域微调过的嵌入模型效果更好。引入重排序Re-ranking先用向量检索出Top 20个候选再用一个更精细的交叉编码器模型对它们进行重排序选出Top 3给LLM准确性会大幅提升。优化查询对用户原始问题进行改写或扩展Query Expansion再用于检索。Token超限错误当对话历史或检索的上下文太长时会超过LLM的上下文窗口。解决为记忆模块实现更智能的截断或总结策略如ConversationSummaryMemory。在RAG中限制检索返回的文本块数量k和每个块的大小chunk_size。精确计算token使用tiktoken等库并在构造提示词前主动截断。6.2 从“最小版”到“可用版”的进阶路径我们这个最小版实现了核心思想但要用于生产还需要在很多方面加固和扩展健壮性错误处理为LLM API调用、工具执行添加更完善的异常捕获和重试机制。解析鲁棒性用更灵活的解析器如基于语法替代简单的正则匹配提高对LLM输出格式波动的容忍度。超时控制为LLM调用和工具执行设置超时防止卡死。功能扩展更多工具实现文件读写、数据库查询、代码执行等工具。多模态集成视觉模型让Agent能“看”图片并描述或分析。复杂工作流引入类似LangGraph的图编排支持循环、分支、并行等复杂Agent逻辑。记忆升级实现向量记忆、摘要记忆让Agent拥有更强大的长期记忆能力。工程化配置化将模型参数、工具列表、提示词模板等外置到配置文件如YAML。可观测性加入详细的日志记录追踪每一次LLM调用、工具执行和token消耗便于调试和成本分析。异步化使用asyncio将LLM调用和I/O密集型工具改为异步提升整体响应速度。通过这个从零构建的过程你应该对LangChain这类框架所解决的问题、其内部的核心抽象和运行机制有了更深刻的理解。下次当你再使用成熟的框架时你会更清楚每一行配置代码背后发生了什么也能在遇到问题时更快地定位到是提示词、工具、记忆还是Agent逻辑本身的问题。这才是“造轮子”最大的价值——不是为了替代而是为了透彻的理解。