如何构建自定义代码 RAG

from lancedb.pydantic import LanceModel, Vectorfrom lancedb.embeddings import get_registrydb = lancedb.connect("/tmp/db")func = get_registry().get("openai").create(    name="voyage-code-3",    base_url="https://api.voyageai.com/v1/",    api_key=os.environ["VOYAGE_API_KEY"],)class CodeChunks(LanceModel):    filename: str    text: str = func.SourceField()    # 1024 is the default dimension for `voyage-code-3`: https://docs.voyageai.com/docs/embeddings#model-choices    vector: Vector(1024) = func.VectorField()table = db.create_table("code_chunks", schema=CodeChunks, mode="overwrite")table.add([    {"text": "print('hello world!')", filename: "hello.py"},    {"text": "print('goodbye world!')", filename: "goodbye.py"}])query = "greetings"actual = table.search(query).limit(1).to_pydantic(CodeChunks)[0]print(actual.text)

"""Custom RAG MCP server for code retrieval"""
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import lancedb

# Initialize your vector database connection
db = lancedb.connect("/path/to/your/db")
table = db.open_table("code_chunks")

app = Server("custom-rag-server")

@app.tool()
async def search_codebase(query: str, limit: int = 10) -> list[TextContent]:
    """
    Search the codebase using vector similarity.
    
    Args:
        query: The search query
        limit: Maximum number of results to return
    """
    # Query your vector database
    results = table.search(query).limit(limit).to_list()
    
    # Format results for Continue
    formatted_results = []
    for result in results:
        formatted_results.append(TextContent(
            type="text",
            text=f"File: {result['filename']}\n\n{result['text']}"
        ))
    
    return formatted_results

@app.tool()
async def get_file_context(filename: str) -> list[TextContent]:
    """
    Get all chunks from a specific file.
    
    Args:
        filename: The name of the file to retrieve
    """
    results = table.where(f"filename = '{filename}'").to_list()
    
    return [TextContent(
        type="text",
        text="\n".join([r['text'] for r in results])
    )]

if __name__ == "__main__":
    stdio_server(app).run()

mcpServers:
  - name: custom-rag
    command: python
    args:
      - /path/to/your/mcp_server.py
    env:
      VOYAGE_API_KEY: ${VOYAGE_API_KEY}

{
  "mcpServers": [
    {
      "name": "custom-rag",
      "command": "python",
      "args": ["/path/to/your/mcp_server.py"],
      "env": {
        "VOYAGE_API_KEY": "${VOYAGE_API_KEY}"
      }
    }
  ]
}