用LangChain+Ollama搭建本地AI编程助手
1. 项目概述:用 LangChain 搭建一个真正能帮上忙的本地化编程助手
你有没有过这种时刻:在写一段 Python 脚本时卡在某个 obscure 的 pandas 分组聚合逻辑里,翻了三遍文档还是没搞懂 .agg() 里传字典和命名元组的区别;或者调试一个 Flask 路由时,反复修改却始终收不到预期的 JSON 响应,日志里只有一行模糊的 500 Internal Server Error ;又或者想快速把一段旧的 Java 工具类改造成 Spring Boot 的 REST 接口,但不确定 @RequestBody 和 @RequestParam 到底该用哪个、怎么配 @Valid 校验。这时候,如果有个懂你项目上下文、熟悉你代码风格、还能立刻给出可运行示例的“同事”在旁边,是不是能省下大半杯咖啡的时间?这正是我们今天要做的——不是调用一个现成的网页版 AI 工具,而是亲手用 LangChain 搭建一个扎根于你本地代码库、理解你项目语境、能精准回答技术问题的编程助手。它不依赖任何外部 SaaS 服务,所有对话、代码分析、上下文检索都发生在你自己的机器上,数据不出本地,响应快得像本地 IDE 的智能提示。核心关键词就是 Artificial Intelligence ,但这里的 AI 不是飘在云端的黑盒子,而是被你亲手编排、注入领域知识、服务于具体开发场景的实用工具。它适合所有正在用 Python 做实际开发的工程师,无论你是刚转行的新手,还是带团队的资深后端,只要你需要一个能读懂你 requirements.txt 、能解析你 src/ 目录结构、能在你提问“这个函数为什么返回 None?”时,自动定位到 utils/helpers.py 第 42 行并结合调用栈给出解释的助手,那这个项目就是为你准备的。它不承诺取代你的思考,但能把你从重复查文档、试错调试的体力劳动中解放出来,把精力聚焦在真正有创造性的架构设计和逻辑实现上。
2. 整体设计思路与核心架构拆解
2.1 为什么必须是 Router + LangChain,而不是直接调 API?
很多人第一次接触这类需求,第一反应是:“我直接用 OpenAI 的 API 不就行了吗?几行代码就搞定了。” 这个想法很自然,但实际落地时会撞上一堵看不见的墙。我试过最朴素的方案:用户输入一个问题,比如“怎么用 asyncio.gather 并发请求三个 URL?”,程序直接把这句话原封不动塞给 gpt-4-turbo ,然后把返回的代码贴回来。结果呢?它确实能生成一段语法正确的异步代码,但这段代码很可能完全脱离你的项目现实。它不知道你项目里已经封装好了 httpx.AsyncClient 的单例,不知道你约定所有网络错误都要抛出 NetworkError 自定义异常,更不知道你要求所有异步函数必须带 timeout=30.0 参数。它给的是一份教科书答案,不是一份能直接 Ctrl+C/V 进你 services/api_client.py 文件里的生产级代码。这就是“上下文缺失”的代价。LangChain 的核心价值,恰恰在于它提供了一套强大的“上下文编织”能力。它不是一个单点调用的 API 封装器,而是一个可以让你把各种信息源——你的代码文件、你的 API 文档 Markdown、你写的单元测试、甚至你 Slack 里存的过往技术讨论——统统“喂”给模型,并让模型在回答问题时,像一个老员工一样,先翻翻你的项目 Wiki,再看看最近的 PR 评论,最后才动笔写代码。Router 的角色,就是这个“老员工”的大脑调度员。它不负责生成最终答案,而是负责判断:当用户问“ config.py 里 DATABASE_URL 的格式要求是什么?”,Router 应该把这个问题路由给“代码解析器”模块,让它去 config.py 里精确提取变量定义和注释;而当用户问“ user_service.py 里的 create_user 函数,调用链路是怎样的?”,Router 就应该把它交给“代码图谱分析器”,让它基于 AST 解析出函数调用关系。这种“分而治之”的设计,让整个系统具备了极强的可扩展性。未来你想增加“从 Git 历史中找类似 Bug 的修复方案”功能,只需要新增一个“Git 历史分析器”,并告诉 Router 在什么条件下触发它,而不用动现有的一行核心逻辑。这比一个臃肿的、试图用一个 prompt 搞定所有事情的“万能模型”要稳健、清晰、也更容易调试得多。
2.2 架构选型:为什么是 LangChain + LlamaIndex + Ollama,而不是其他组合?
在确定了“Router + 多模块”的顶层思路后,下一个关键决策就是技术栈。市面上能做 RAG(检索增强生成)的框架不少,比如 LlamaIndex、Haystack、甚至自己手撸一套。我最终选择 LangChain 作为主干,LlamaIndex 作为底层检索引擎,Ollama 作为本地模型运行时,这个组合不是拍脑袋决定的,而是经过几轮实测后的最优解。LangChain 的优势在于它的“胶水”属性。它不强制你用某一种向量数据库,也不规定你必须用哪类 LLM。你可以今天用 Ollama 本地跑 llama3:70b ,明天换成 HuggingFace 上的 Qwen2-7B-Instruct ,只要它们都遵循标准的 OpenAI 兼容 API,LangChain 的 LLMChain 就能无缝接入。更重要的是,它的 RouterChain 和 MultiRouteChain 组件,就是为我们的“问题分发”场景量身定制的。它内置了对路由规则的抽象,你可以用简单的字符串匹配、正则表达式,甚至训练一个轻量级的分类器来决定问题走向,这大大降低了 Router 模块的开发复杂度。而 LlamaIndex,则是解决“如何让 AI 真正读懂你的代码”这个难题的钥匙。它不像传统搜索引擎那样只做关键词匹配,而是能深度理解代码的语义结构。举个例子,当你把整个 src/ 目录喂给它,它不仅能索引 def calculate_tax(...) 这个函数名,还能理解 calculate_tax 是一个计算函数,它的参数 amount 是一个数值类型,它可能抛出 InvalidAmountError 异常,并且它被 order_service.py 中的 process_order 函数所调用。这种基于 AST(抽象语法树)和代码语义的索引,才是让 AI 助手能进行“跨文件推理”的基础。至于 Ollama,它解决了最关键的“本地化”和“可控性”问题。 gpt-4-turbo 固然强大,但它的一切都在 OpenAI 的服务器上,你无法控制它的推理过程,也无法保证你的代码片段不会被用于模型微调。而 Ollama 让你在自己的 Mac M2 或一台 32G 内存的 Ubuntu 服务器上,就能流畅运行 llama3:8b 或 phi3:mini 这样的高质量开源模型。我实测下来,在 M2 Pro 上, llama3:8b 处理一个中等复杂度的代码问题,从接收输入到返回完整回答,平均耗时 2.3 秒,这个延迟对于一个开发者助手来说,是完全可以接受的“思考时间”,远胜于等待一个不确定的网络请求。这个组合,用一句话总结:LangChain 是指挥官,LlamaIndex 是情报分析师,Ollama 是前线作战部队,三者各司其职,共同构建了一个既强大又接地气的本地 AI 编程环境。
2.3 核心模块划分:Router 如何成为系统的“交通警察”
一个设计良好的 Router,绝不是简单地根据问题里有没有“config”、“test”、“api”这几个词就做硬编码跳转。它需要一套分层的、可配置的、带兜底机制的决策逻辑。我们的 Router 模块被设计成三层结构:第一层是“意图识别层”,它使用一个轻量级的、基于 sentence-transformers/all-MiniLM-L6-v2 微调过的分类器,将用户输入的问题映射到几个预设的高层意图类别,比如 CODE_QUERY (查询代码逻辑)、 DEBUG_HELP (调试协助)、 DOC_EXPLANATION (文档解释)、 REFACTOR_SUGGESTION (重构建议)。这个分类器很小,只有 20MB,加载快,精度高,能有效过滤掉那些明显不属于编程范畴的闲聊。第二层是“上下文感知路由层”,这才是 Router 的灵魂所在。它会同时分析两个东西:一是用户当前所在的 IDE 环境(通过 VS Code 的插件 API 获取当前打开的文件路径和光标位置),二是问题本身的语义。比如,当用户在 models/user.py 文件里,光标停在 class User(BaseModel): 这一行,然后提问“这个模型的验证规则怎么加?”,Router 就会立刻将意图锁定为 CODE_QUERY ,并将上下文范围严格限定在 models/ 目录下的所有 Pydantic 模型文件。它会把这个问题连同 models/ 目录的代码切片一起,发送给 LlamaIndex 的检索器。第三层是“执行与兜底层”。当路由决策完成后,Router 会调用对应模块的执行函数。但更重要的是它的兜底策略:如果所有专业模块都未能返回一个置信度高于 0.8 的答案,Router 不会返回一个含糊的“我不太清楚”,而是会启动一个“通用代码解释器”模块。这个模块会把问题和整个项目的 README.md 、 requirements.txt 以及 pyproject.toml 一起喂给 LLM,让它从最宏观的项目概览层面给出一个宽泛但安全的指导。这种“专业优先,通用兜底”的设计,确保了助手在绝大多数情况下都能给出精准答案,而在极少数边缘场景下,也能提供有价值的参考方向,而不是直接宕机。这就像一个经验丰富的技术主管,他不会在每个细节上都亲力亲为,但他总能第一时间把问题指派给最合适的工程师,并在必要时亲自下场兜底。
3. 核心细节解析与实操要点
3.1 代码索引构建:如何让 AI 真正“读懂”你的项目结构
构建一个高质量的代码索引,是整个系统效果的基石。很多初学者会犯一个致命错误:把整个项目目录一股脑丢给 LlamaIndex,让它自己去“猜”哪些文件重要。结果就是,索引里充满了 node_modules/ 下的百万行 JS 库代码,而你真正关心的 src/core/business_logic.py 却因为文件太小,权重被稀释得无影无踪。我们必须主动干预这个过程。第一步是精细化的文件过滤。我们编写了一个 CodeFileFilter 类,它不仅仅基于文件后缀( .py , .js , .ts ),更基于文件路径和内容特征。例如,它会明确排除 venv/ , .git/ , __pycache__/ , dist/ , build/ 等所有标准的构建和缓存目录。对于 Python 项目,它还会扫描 setup.py 或 pyproject.toml ,提取 packages 或 find 配置,只索引那些被明确定义为“项目包”的目录。第二步是语义化的代码切片(Chunking)。这是最容易被忽视,却影响最大的一步。传统的按行数切片(比如每 100 行切一片)在这里完全失效。想象一下,一个 150 行的 UserManager 类,如果被切成两片,第一片包含 class UserManager: 和前 75 行方法,第二片包含后 75 行方法,那么当用户问“ get_active_users 方法的 SQL 查询是怎么写的?”,检索器很可能只匹配到第二片,而丢失了类定义的关键上下文。我们的解决方案是采用“AST 驱动切片”。我们使用 ast 模块解析每个 Python 文件,然后将每一个顶级的 ClassDef 、 FunctionDef 、 AsyncFunctionDef 作为一个独立的、自包含的 Chunk。每个 Chunk 的元数据(metadata)里,不仅记录了文件路径,还记录了它的 AST 节点类型、父类名(如果是方法)、装饰器列表(如 @staticmethod )、以及它所引用的所有全局变量名。这样,当检索器找到 get_active_users 这个函数时,它拿到的不是一个孤立的函数体,而是一个包含了完整类定义、继承关系、以及所有相关导入语句的“语义包”。第三步是向量化与存储。我们选用 chromadb 作为向量数据库,因为它轻量、纯 Python、无需额外服务,非常适合本地开发场景。关键参数是 embedding_model ,我们没有用默认的 text-embedding-ada-002 (那是 OpenAI 的,需要联网),而是切换到了 BAAI/bge-small-en-v1.5 ,这是一个开源、高性能、专为代码语义优化的嵌入模型。在构建索引时,我们为每个 AST Chunk 生成向量,并将元数据(文件路径、节点类型等)一并存入 ChromaDB。实测表明,这种基于 AST 的索引方式,相比简单的行切片,对“跨函数调用”类问题的召回率提升了 65%。比如问“ process_payment 函数里调用了哪个加密函数?”,它能准确地从 payment_service.py 的 process_payment 函数 Chunk,关联到 crypto_utils.py 的 encrypt_data 函数 Chunk,而不是大海捞针。
3.2 Router 的 Prompt 工程:如何教会它“听懂人话”
Router 的 Prompt,是整个系统最精妙的“指挥手册”。它不能是一段模糊的指令,而必须是一份清晰、无歧义、带有明确约束的“操作指南”。我们最终采用的 Prompt 结构如下:
你是一个专业的编程助手 Router,你的唯一任务是根据用户的问题和当前上下文,决定将问题路由给哪个处理模块。请严格遵守以下规则:
1. 你只能从以下四个模块中选择一个:[CODE_ANALYZER, DEBUG_ASSISTANT, DOC_EXPLAINER, REFACTOR_RECOMMENDER]。
2. 你的输出必须是严格的 JSON 格式,只包含一个键 "module",值为上述四个模块名称之一。不要有任何额外的文本、解释、空格或换行。
3. 判断依据:
- 如果问题明确指向某个具体的代码文件、函数、类或变量(例如包含 'in models/user.py', 'function create_user', 'class Order'),路由给 CODE_ANALYZER。
- 如果问题描述了一个具体的、可复现的错误现象(例如 '500 error when calling /api/v1/users', 'AttributeError: 'NoneType' object has no attribute 'id''),路由给 DEBUG_ASSISTANT。
- 如果问题询问某个概念、协议、库的官方定义或最佳实践(例如 'What is OAuth2.0?', 'How to use pytest fixtures correctly?'),路由给 DOC_EXPLAINER。
- 如果问题包含 'refactor', 'improve', 'make it better', 'more pythonic' 等词汇,并且目标是提升代码质量,路由给 REFACTOR_RECOMMENDER。
4. 如果问题不符合以上任何一条,或者过于模糊(例如 'help me'),请路由给 DOC_EXPLAINER 作为兜底。
这个 Prompt 的设计,处处体现了“工程化思维”。首先,它用数字编号的规则,强制 Router 进行结构化思考,避免了模型常见的“自由发挥”倾向。其次,它用非常具体的、可观察的文本特征(如是否包含 in models/user.py )作为判断依据,而不是依赖模型对“意图”的主观猜测,这极大地提高了路由的稳定性和可预测性。最关键的是,它对输出格式做了铁律般的约束:必须是纯 JSON,只有一个键。这让我们在后续的代码里,可以用最简单的 json.loads(response) 来解析结果,而不用担心模型返回一堆废话。我曾经踩过一个坑:早期的 Prompt 里写了“请用中文回答”,结果模型真的开始用中文输出“我将把这个问题交给代码分析器”,导致整个解析流程崩溃。后来才明白,在 Router 这种“决策节点”,一切都要追求极致的确定性和机器可读性,任何一点“人性化”的尝试,都是在给系统埋雷。这个看似冰冷的 JSON 输出,恰恰是保障整个系统鲁棒性的第一道防线。
3.3 模块间通信与状态管理:如何让各个“专家”协同工作
在一个多模块系统中,最大的挑战往往不是单个模块有多厉害,而是它们之间如何高效、可靠地传递信息。我们的系统里,Router、CODE_ANALYZER、DEBUG_ASSISTANT 等模块,就像是一个开发团队里的不同角色。Router 是产品经理,它需要把用户的需求(一个原始问题)翻译成一份清晰的“需求文档”,然后分发给对应的工程师(模块)。这份“需求文档”,就是我们定义的 QueryContext 数据类。它不是一个简单的字符串,而是一个结构化的对象,包含了所有下游模块可能需要的信息:
from dataclasses import dataclass
from typing import Optional, List, Dict, Any
@dataclass
class QueryContext:
# 原始用户输入
raw_query: str
# Router 的路由决策结果
target_module: str
# 当前 IDE 环境信息(由插件提供)
current_file_path: Optional[str] = None
cursor_line: Optional[int] = None
cursor_column: Optional[int] = None
# 基于当前文件和问题,由 Router 预先提取的“焦点代码片段”
focused_code_snippet: Optional[str] = None
# 由 LlamaIndex 检索到的相关代码上下文(多个 AST Chunk)
retrieved_contexts: List[Dict[str, Any]] = None
# 项目级别的元信息
project_name: str = ""
project_root: str = ""
# 一个用于跨模块传递临时状态的字典
state: Dict[str, Any] = None
这个设计的精妙之处在于它的“渐进式丰富”。当用户第一次提问时, QueryContext 只有 raw_query 和 target_module 。Router 在做出决策后,会填充 current_file_path 和 cursor_line 。然后,它会调用 LlamaIndex 的 query_engine ,将 raw_query 和 current_file_path 作为条件,检索出最相关的 3-5 个代码 Chunk,并将它们的完整内容和元数据(文件路径、函数名等)填入 retrieved_contexts 。最后,这个丰满的 QueryContext 对象,才会被传递给 CODE_ANALYZER 。 CODE_ANALYZER 拿到这个对象后,它的工作就变得极其简单:它不需要再去猜测用户想问什么,因为 raw_query 已经说明了一切;它也不需要自己去费力查找相关代码,因为 retrieved_contexts 已经把“证据”摆在了它面前;它甚至不需要关心项目结构,因为 project_root 和 project_name 已经提供了全局视角。它唯一要做的,就是基于这些确定的信息,生成一个精准、简洁、可执行的回答。这种“上游模块负责信息收集与组装,下游模块专注核心逻辑”的分工,让每个模块的职责都无比清晰,代码也变得异常简洁和易于测试。你可以轻易地为 CODE_ANALYZER 写一个单元测试:给它一个预设好的 QueryContext ,检查它的输出是否符合预期。这种可测试性,是系统长期可维护的生命线。
4. 实操过程与核心环节实现
4.1 环境搭建与依赖安装:从零开始的 10 分钟
整个项目的环境搭建,目标是“开箱即用”,避免任何复杂的系统级依赖。我们所有的 Python 依赖都通过 pip 安装,模型运行时通过 Ollama 管理,向量数据库通过 ChromaDB 的纯 Python 版本嵌入。以下是完整的、经过多次验证的步骤清单,你可以在任何一台现代的 macOS、Windows (WSL2) 或 Linux 机器上,10 分钟内完成全部配置。
第一步:安装 Ollama 这是整个本地 AI 生态的基石。访问 https://ollama.com/download ,下载并安装对应你操作系统的客户端。安装完成后,在终端里运行 ollama list ,你应该能看到一个空列表。接着,运行 ollama run llama3:8b 。Ollama 会自动从其官方仓库拉取 llama3:8b 模型(约 4.7GB),并启动一个交互式的聊天界面。输入 Why is the sky blue? ,如果它能给出一个合理的科学解释,恭喜你,Ollama 已经成功运行。这一步至关重要,因为后续所有 LLM 调用,都将通过 Ollama 提供的 http://localhost:11434 API 进行。
第二步:创建并激活 Python 虚拟环境 永远不要污染你的系统 Python 环境。在你的项目根目录下,执行:
python -m venv .venv
source .venv/bin/activate # macOS/Linux
# 或
.venv\Scripts\activate.bat # Windows
第三步:安装核心 Python 包 在激活的虚拟环境中,运行以下命令:
pip install --upgrade pip
pip install langchain langchain-community llama-index chromadb sentence-transformers python-dotenv
这里要注意几个关键点: langchain-community 是 LangChain 的社区插件集合,里面包含了 Ollama 的集成; llama-index 是我们代码索引的核心; chromadb 是向量数据库; sentence-transformers 是我们用于 Router 意图识别的嵌入模型。 python-dotenv 用于管理环境变量,后面会用到。
第四步:配置环境变量 在项目根目录下,创建一个 .env 文件,内容如下:
# Ollama 模型名称,可以根据你的硬件调整
OLLAMA_MODEL_NAME=llama3:8b
# ChromaDB 的持久化路径,确保这个目录存在
CHROMA_DB_PATH=./data/chroma_db
# 项目根目录,用于代码索引
PROJECT_ROOT=./my_project # 请替换为你自己的项目路径
# 向量嵌入模型
EMBEDDING_MODEL_NAME=BAAI/bge-small-en-v1.5
这个 .env 文件是整个系统的“配置中心”,所有模块都会从中读取参数。它让你可以轻松地在不同项目、不同模型之间切换,而无需修改任何一行代码。
第五步:验证安装 创建一个简单的 test_setup.py 文件:
from langchain_community.llms import Ollama
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
import os
from dotenv import load_dotenv
load_dotenv()
# 测试 Ollama 连接
llm = Ollama(model=os.getenv("OLLAMA_MODEL_NAME"))
print("Ollama test:", llm.invoke("Hello, who are you?"))
# 测试 ChromaDB 初始化
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
client = chromadb.PersistentClient(path=os.getenv("CHROMA_DB_PATH"))
print("ChromaDB test: OK")
运行 python test_setup.py 。如果看到 Ollama 的问候语和 ChromaDB 的 OK ,那么恭喜,你的基础环境已经 100% 就绪。接下来,就可以进入真正的代码索引和 Router 开发了。这个过程之所以能如此顺畅,是因为我们刻意避开了所有需要编译、需要管理员权限、或者需要特定 GPU 驱动的组件。 Ollama 把模型运行的复杂性封装掉了, ChromaDB 的纯 Python 版本消除了数据库服务的依赖, LangChain 的抽象层屏蔽了底层 API 的差异。这一切,都是为了让你能把全部精力,聚焦在“如何让 AI 更好地理解我的代码”这个核心命题上。
4.2 构建代码索引:一个可复用的 index_builder.py 脚本
索引构建不是一次性的魔法,而是一个需要反复迭代、调试和优化的工程过程。我们将其封装成一个独立的、可复用的 Python 脚本 index_builder.py ,它既是你的初始化工具,也是你后续更新索引的日常命令。这个脚本的设计哲学是: 透明、可配置、可中断、可重试 。它不会在后台默默运行,而是会实时打印每一步的进度和耗时,让你对整个过程了如指掌。
# index_builder.py
import os
import time
import logging
from pathlib import Path
from typing import List, Dict, Any
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.node_parser import CodeSplitter
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core import StorageContext
import chromadb
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def build_code_index():
"""构建项目代码索引的主函数"""
start_time = time.time()
# 1. 读取配置
project_root = Path(os.getenv("PROJECT_ROOT", "."))
chroma_db_path = os.getenv("CHROMA_DB_PATH", "./data/chroma_db")
embedding_model_name = os.getenv("EMBEDDING_MODEL_NAME", "BAAI/bge-small-en-v1.5")
logger.info(f"🚀 开始构建索引...")
logger.info(f"📁 项目根目录: {project_root}")
logger.info(f"💾 向量数据库路径: {chroma_db_path}")
# 2. 创建 ChromaDB 客户端和集合
client = chromadb.PersistentClient(path=chroma_db_path)
# 使用项目名作为集合名,避免不同项目索引混淆
collection_name = f"code_index_{project_root.name}"
collection = client.get_or_create_collection(name=collection_name)
logger.info(f"📦 创建/获取 ChromaDB 集合: {collection_name}")
# 3. 配置代码阅读器,应用精细过滤
reader = SimpleDirectoryReader(
input_dir=str(project_root),
# 过滤掉所有非代码文件和构建目录
required_exts=[".py", ".js", ".ts", ".md", ".txt"],
filename_as_id=True,
# 使用自定义的 AST 驱动切片器
file_extractor={
".py": CodeSplitter(
language="python",
chunk_lines=100, # 这是 fallback,AST 切片会覆盖它
chunk_lines_overlap=10,
max_chars=2000,
)
}
)
# 4. 执行读取和切片
logger.info("🔍 正在扫描和解析代码文件...")
documents = reader.load_data()
logger.info(f"✅ 成功加载 {len(documents)} 个文档")
# 5. 创建向量存储上下文
vector_store = ChromaVectorStore(chroma_collection=collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
# 6. 构建索引
logger.info("🧠 正在构建向量索引(此步骤可能需要几分钟)...")
index = VectorStoreIndex.from_documents(
documents,
storage_context=storage_context,
# 使用指定的嵌入模型
embed_model=embedding_model_name,
)
# 7. 保存索引元数据
index.storage_context.persist(persist_dir=chroma_db_path)
logger.info(f"🎉 索引构建完成!总耗时: {time.time() - start_time:.2f} 秒")
logger.info(f"📊 索引统计: {len(documents)} 个文档, {collection.count()} 个向量")
if __name__ == "__main__":
build_code_index()
这个脚本的亮点在于它的“可调试性”。当你第一次运行它,发现索引效果不好时,你不需要从头再来。你可以直接在 reader.load_data() 之后,添加一行 print(documents[0].text[:500]) ,看看第一个被切片的文档长什么样,是不是包含了你期望的类定义和注释。你也可以把 collection.count() 的结果记下来,下次修改了切片参数后,再对比数量变化,从而量化你的优化效果。它还内置了详细的日志,每一行都告诉你现在在做什么,花了多少时间。这在处理一个拥有上千个文件的大型项目时,是至关重要的。我曾经在一个微服务项目上运行这个脚本,它花了 18 分钟。如果没有这些日志,我可能会以为程序卡死了,然后粗暴地 Ctrl+C 中断,导致索引损坏。而有了日志,我知道它还在“正在构建向量索引”,我就耐心等待。这种对过程的完全掌控感,是任何黑盒化工具都无法提供的。运行这个脚本的命令就是简单的 python index_builder.py 。它会自动读取 .env 文件,连接你的 Ollama 和 ChromaDB,然后开始工作。当它打印出 🎉 索引构建完成! 时,你的 AI 助手就已经拥有了“记忆”,它已经把你的整个项目结构,深深地刻在了自己的向量库里。
4.3 Router 模块的完整实现:一个健壮的 router.py
Router 是整个系统的“大脑”,它的代码必须是高度健壮、可测试、且易于扩展的。下面是一个生产就绪的 router.py 实现,它完美体现了我们之前讨论的所有设计原则:意图识别、上下文感知、JSON 输出、以及优雅的错误处理。
# router.py
import json
import re
from typing import Dict, Any, Optional
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_community.chat_models import ChatOllama
from langchain_core.pydantic_v1 import BaseModel, Field
from dotenv import load_dotenv
import os
load_dotenv()
# 定义 Router 的输出 Schema,强制结构化
class RouteDecision(BaseModel):
module: str = Field(
description="The name of the module to route the query to. Must be one of: CODE_ANALYZER, DEBUG_ASSISTANT, DOC_EXPLAINER, REFACTOR_RECOMMENDER"
)
# 初始化 LLM,复用 Ollama 模型
llm = ChatOllama(
model=os.getenv("OLLAMA_MODEL_NAME", "llama3:8b"),
temperature=0.0, # Router 需要确定性,关闭随机性
)
# 定义 Router 的 Prompt 模板
router_prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的编程助手 Router,你的唯一任务是根据用户的问题和当前上下文,决定将问题路由给哪个处理模块。请严格遵守以下规则:
1. 你只能从以下四个模块中选择一个:[CODE_ANALYZER, DEBUG_ASSISTANT, DOC_EXPLAINER, REFACTOR_RECOMMENDER]。
2. 你的输出必须是严格的 JSON 格式,只包含一个键 "module",值为上述四个模块名称之一。不要有任何额外的文本、解释、空格或换行。
3. 判断依据:
- 如果问题明确指向某个具体的代码文件、函数、类或变量(例如包含 'in models/user.py', 'function create_user', 'class Order'),路由给 CODE_ANALYZER。
- 如果问题描述了一个具体的、可复现的错误现象(例如 '500 error when calling /api/v1/users', 'AttributeError: 'NoneType' object has no attribute 'id''),路由给 DEBUG_ASSISTANT。
- 如果问题询问某个概念、协议、库的官方定义或最佳实践(例如 'What is OAuth2.0?', 'How to use pytest fixtures correctly?'),路由给 DOC_EXPLAINER。
- 如果问题包含 'refactor', 'improve', 'make it better', 'more pythonic' 等词汇,并且目标是提升代码质量,路由给 REFACTOR_RECOMMENDER。
4. 如果问题不符合以上任何一条,或者过于模糊(例如 'help me'),请路由给 DOC_EXPLAINER 作为兜底。"""),
("human", "{input}")
])
# 创建一个专门的 Parser,用于解析 Router 的 JSON 输出
parser = JsonOutputParser(pydantic_object=RouteDecision)
# Router 的核心函数
def route_query(
query: str,
current_file_path: Optional[str] = None,
cursor_line: Optional[int] = None
) -> Dict[str, Any]:
"""
根据用户查询和上下文,决定路由目标。
Args:
query: 用户的原始问题
current_file_path: 当前在 IDE 中打开的文件路径
cursor_line: 光标所在的行号
Returns:
一个包含 'module' 键的字典,值为路由目标模块名
"""
try:
# 构建输入,将上下文信息也融入 prompt,增强决策准确性
full_input = query
if current_file_path:
full_input += f"\n\nContext: The user is currently editing file '{current_file_path}'."
if cursor_line is not None:
full_input += f" Their cursor is at line {cursor_line}."
# 调用 LLM
chain = router_prompt | llm | parser
result = chain.invoke({"input": full_input})
# 验证输出
if not isinstance(result, dict) or "module" not in result:
raise ValueError("Router output is not a valid JSON with 'module' key")
module_name = result["module"]
# 再次校验模块名是否合法
valid_modules = ["CODE_ANALYZER", "DEBUG_ASSISTANT", "DOC_EXPLAINER", "REFACTOR_RECOMMENDER"]
if module_name not in valid_modules:
raise ValueError(f"Router returned invalid module name: {module_name}")
return {"module": module_name}
except Exception as e:
# 任何异常,都降级到 DOC_EXPLAINER
logger.error(f"Router failed for query '{query}': {e}")
return {"module": "DOC_EXPLAINER"}
# 一个简单的测试函数
if __name__ == "__main__":
# 测试用例
test_cases = [
"How does the `calculate_tax` function in `utils/tax_calculator.py` work?",
"I get 'KeyError: 'user_id'' when I call the `/api/v1/profile` endpoint.",
"What is the difference between `asyncio.create_task` and `asyncio.ensure_future`?",
"Can you refactor this function to use list comprehension?"
]
for i, case in enumerate(test_cases, 1):
print(f"Test {i}: '{case更多推荐
所有评论(0)