Skip to main content

· 14 min read
Shixu

过去几年里我一直使用Java。最近,用Go建立了一个小项目,然而 Go 生态系统中依赖注入(DI)功能缺乏让我震惊。于是我决定尝试使用 Uber 的 dig 库来构建我的项目,期间感触颇深。

我发现 DI 帮助我解决了之前在 Go 应用程序中遇到的很多问题 - 过度使用 init 函数,滥用全局变量和复杂的应用程序设置等。

在这篇文章中,我将介绍 DI ,然后在使用 DI 框架(通过 dig 库)前后写一些例子做对比。

DI 的简要概述

依赖注入是指你的组件(通常在 go 中是 struct )在创建时,就应该获取它们依赖关系的一种思想。这与那些组件在初始化过程中,就建立自身依赖关系的反关联模式不同 。我们来看一个例子。

假设你构造Server 需要 Config 结构体。一种方法是在初始化期间 Server 构建 Config

type Server struct {
config *Config
}
func New() *Server {
return &Server{
config: buildMyConfigSomehow(),
}
}

看起来很方便。调用者甚至不必知道 Server 需要访问 Config 。这些都被我们的函数隐藏起来了。

然而,这存在一些缺点。首先,如果我们想要改变我们 Config 的构建方式,我们不得不改变所有调用构建代码的地方。例如,假设我们的 buildMyConfigSomehow 函数现在需要一个参数。每个调用处都需要访问该参数并需要将其传递给构造函数。

此外,这使得实现 Config 函数变得十分麻烦,我们得以某种方法进入 new 函数的内部,并创建Config

这是 DI 方式:

type Server struct {
config *Config
}
func New(config *Config) *Server {
return &Server{
config: config,
}
}

现在我们将 ServerConfig 分离 。我们可以根据自己的逻辑创造 Config 然后将结果传递给 New 函数。

此外,如果 Config 是一个接口,这为我们提供了一个简单的模拟途径 。只要 New 实现了我们的接口,就可以传递任何我们想要的东西。这使得测试实现了 Config 接口的 Server 很简单。

令人痛苦的是在创建 server 之前手动创建 config 。我们在这里创建了一个依赖关系 – 因为 server 依赖 Config, 所以需要首先创建 Config 。在真正的应用程序中,这些依赖会变得更加复杂,这会导致构建应用程序完成其工作所需的组件间的复杂逻辑 。

这是 DI 框架可以提供帮助的地方。 DI 框架通常提供两个功能:

  1. “提供”新组件。简而言之,这告诉 DI 框架一旦你有这些组件,还需要其他什么组件(依赖关系)以及如何去构建。
  2. “检索”构建组件。

DI 框架通常基于您告诉它的“ providers ”构建依赖图并确定如何构建对象。这在没有具体例子的情况下很难理解,所以让我们来看一个中等大小的例子。

示例程序

我们来看http服务器端的代码 :客户端以 GET 方式请求 /people 路径时并返回 JSON 。我们将一步一步呈现代码,为简单起见,它们都存在于同一个包中(main)。请勿在真正的 Go 程序中执行此操作。可以在此处找到此示例的完整代码。

首先,让我们看看我们的 Person 。仅有一些被 JSON 标签标记的属性。

type Person struct {
Id int `json:"id"`
Name string `json:"name"`
Age int `json:"age"`
}

PersonIdNameAge

接下来让我们看看 Config 。与 Person 类似,它没有依赖关系。与 Person 不同的是,我们将提供构造函数。

type Config struct {
Enabled bool
DatabasePath string
Port string
}
func NewConfig() *Config {
return &Config{
Enabled: true,
DatabasePath: "./example.db",
Port: "8000",
}
}

Enabled 表示程序是否返回真实数据。DatabasePath 表示数据库的地址(使用 sqlite )。Port 表示服务器运行的端口。

下方函数用来打开数据库连接。它依赖于 Config 并返回 *sql.DB

接下来看看 PersonRepository。此结构负责从数据库中提取数据并反序列化为 Person

type PersonRepository struct {
database *sql.DB
}
func (repository *PersonRepository) FindAll() []*Person {
rows, _ := repository.database.Query(
`SELECT id, name, age FROM people;`
)
defer rows.Close()
people := []*Person{}
for rows.Next() {
var (
id int
name string
age int
)
rows.Scan(&id, &name, &age)
people = append(people, &Person{
Id: id,
Name: name,
Age: age,
})
}
return people
}
func NewPersonRepository(database *sql.DB) *PersonRepository {
return &PersonRepository{database: database}
}

PersonRepository的构建需要数据库连接。它有一个函数FindAll,此函数使用数据库连接信息并返回 Person 列表。

要在 HTTP 服务器和 PersonRepository 之间提供一层,我们需要创建 PersonService

type PersonService struct {
config *Config
repository *PersonRepository
}
func (service *PersonService) FindAll() []*Person {
if service.config.Enabled {
return service.repository.FindAll()
}
return []*Person{}
}
func NewPersonService(config *Config, repository *PersonRepository)
*PersonService {
return &PersonService{config: config, repository: repository}
}

我们的 PersonService 依赖于 ConfigPersonRepository 。它有一个函数 FindAll ,如果启用了应用程序,则会有条件地调用 PersonRepository

最后,我们得到了 Server 。负责运行 HTTP 服务器并将适当的请求委托给 PersonService

type Server struct {
config *Config
personService *PersonService
}
func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/people", s.people)
return mux
}
func (s *Server) Run() {
httpServer := &http.Server{
Addr: ":" + s.config.Port,
Handler: s.Handler(),
}
httpServer.ListenAndServe()
}
func (s *Server) people(w http.ResponseWriter, r *http.Request) {
people := s.personService.FindAll()
bytes, _ := json.Marshal(people)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
w.Write(bytes)
}
func NewServer(config *Config, service *PersonService) *Server {
return &Server{
config: config,
personService: service,
}
}

Server 取决于 PersonServiceConfig

好的,我们了解了系统的所有组件。现在我们该如何在实际中初始化它们并启动我们的系统?

传统的 main()

首先,让我们用传统方式编写 main()

func main() {
config := NewConfig()
db, err := ConnectDatabase(config)
if err != nil {
panic(err)
}
personRepository := NewPersonRepository(db)
personService := NewPersonService(config, personRepository)
server := NewServer(config, personService)
server.Run()
}

首先,我们创建 Config 。然后使用 Config 创建数据库连接。从而创建 PersonRepositoryPersonService 。最后,再创建 Server 并运行它。

这有些复杂。更糟糕的是,随着我们的应用程序的变得复杂, main 的复杂性也将继续增长。每次我们向任何组件添加新的依赖时,都必须通过 main 函数中的排序和逻辑来反映该依赖,以构建该组件。

您可能已经猜到,依赖注入框架可以帮助我们解决这个问题。一起来看看。

创建容器

术语“ 容器(container) ”通常用在 DI 框架中,用于描述添加“ 提供者(providers) ”的内容,并从中请求构建对象。 dig 库用 Provide 函数为我们添加“ providers ”, Invoke 函数用于从容器中检索全部的构建对象。

首先,我们构建一个新容器。

container := dig.New()

现在我们可以添加新的提供者。为此,我们在容器上调用 Provide 函数。它只需要一个参数:一个函数。此函数可以包含任意数量的参数(表示要创建的组件的依赖关系)和一个或两个返回值(表示函数提供的组件以及可选的错误)。

container.Provide(func() *Config {
return NewConfig()
})

上面的代码说“我为容器提供了一种 Config 类型。为了构建它,我不需要任何其他东西。“现在我们已经向容器展示了如何构建 Config 类型,继续使用它来构建其他类型。

container.Provide(func(config *Config) (*sql.DB, error) {
return ConnectDatabase(config)
})

这段代码说“我为容器提供了一种 *sql.DB 类型。为了构建它,我需要一个 Config 。可以选择返回错误。“

在这两种情况下,我们没必要这样写。因为我们已经有了 NewConfigConnectDatabase 函数,我们可以直接使用他们作为容器的提供者。

container.Provide(NewConfig)
container.Provide(ConnectDatabase)

现在,我们可以从之前给容器提供的类型中创建组件。我们使用 Invoke 函数,函数采用单个参数 - 具有任意数量参数的函数。函数的参数是我们希望容器构建的类型。

container.Invoke(func(database *sql.DB) {
// sql.DB is ready to use here
})

容器做了一些非常聪明的东西,如下:

  • 容器认识到我们要求的是构建 *sql.DB
  • 它确定函数 ConnectDatabase 提供该类型
  • 接下来它确定 ConnectDatabase 函数依赖 Config
  • 它找到了 Config 的提供者,也就是 NewConfig
  • NewConfig 没有任何依赖关系,所以它被调用
  • NewConfig 的结果是一个 Config 传递给 ConnectDatabase
  • ConnectionDatabase 的结果是 *sql.DB 被传递给 Invoke

这是容器为我们做的很多工作。事实上,它做的更多。容器很智能,可以构建每种类型有且仅有一个实例。这意味着如果我们在多个地方(比如多个存储库)使用它,我们永远不会意外地创建第二个数据库连接。

较好的 main()写法

现在知道了 dig 容器是如何工作的,让我们用它来构建一个较好的main 。

func BuildContainer() *dig.Container {
container := dig.New()
container.Provide(NewConfig)
container.Provide(ConnectDatabase)
container.Provide(NewPersonRepository)
container.Provide(NewPersonService)
container.Provide(NewServer)
return container
}
func main() {
container := BuildContainer()
err := container.Invoke(func(server *Server) {
server.Run()
})
if err != nil {
panic(err)
}
}

之前唯一没见过的就是 Invoke 的返回值 error 。如果任何提供者使用 Invoke 返回错误,我们调用 Invoke 将停止并返回该错误。

虽然这个例子很小,但应该很容易看出这种方法的一些好处超过了“常规“的 main 。随着应用程序变得越来越大,这些好处变得更加明显。

最重要的好处之一是将组件的创建与其依赖的创建分离。比如说,我们 PersonRepository 现在需要访问 Config 。我们所要做的就是更改 NewPersonRepository 构造函数以包含 Config 作为参数。代码其他任何内容没有发生改变。

其他的好处是没有全局状态,没有调用 init (依赖关系在需要时才创建,只创建一次,不需要容易出错的 init 设置),并且易于测试单个组件。想象一下,在测试中创建容器并要求完整构建对象进行测试。或者,创建一个对象需要所有的依赖。使用 DI ,这些都更容易。

一个值得传播的想法

我相信依赖注入有助于构建更强大和可测试的应用程序。随着这些应用程序体量逐渐增大,尤为明显。 Go 非常适合构建大型应用程序,并且具有很好的 DI 工具 dig 。我相信 Go 社区应该接受 DI 并在更多的应用程序中使用它。

原文:dependency-injection-in-go

· 14 min read
Shixu

原文作者:Tal Sheffer | 来源:qodo.ai blog


最近我们看到了不少很酷的生成式 AI 编程演示,有些甚至会让你觉得,仿佛已经有一个勤奋的 AI Agent 正在疯狂承接 Upwork 上的项目。话虽如此,这些“Upwork 大神”式的 AI,在面对拥有数千个代码仓库、数百万行(大多是遗留)代码的真实企业级代码库时,还是完全不够看的。对于希望采用生成式 AI 的企业开发者来说,上下文感知能力是成功的关键。这正是 Retrieval Augmented Generation (RAG) 技术的用武之地,然而,要将 RAG 落地到大规模代码库中也面临着独特的挑战。

在企业级层面使用 RAG 的首要障碍之一是 scalability(可扩展性)。RAG 模型必须处理海量数据,并应对跨不同仓库的架构复杂性,这使得实现上下文理解变得困难。在这篇博客中,我将分享 qodo(前身为 Codium)如何通过 RAG 方法,在构建以代码质量和完整性为先的生成式 AI 编程平台的同时,弥合 context windows 有限的 LLM 与庞大复杂代码库之间的差距。

将 RAG 应用于大规模代码仓库

RAG 大致可以分为两部分:索引知识库(在我们的例子中是代码库)和检索。对于不断变化的生产环境代码库,索引并不是一次性或周期性的工作。我们需要一个强大的 pipeline 来持续维护最新的索引。下图展示了我们的 ingest pipeline:文件被路由到适当的 splitter 进行 chunking,chunk 会被加上自然语言描述进行增强,然后为每个 chunk 生成 vector embeddings,最后存储在 vector DB 中。

Chunking

对于自然语言文本,Chunking 相对简单——段落(和句子)提供了明显的边界点,可以创建语义上有意义的片段。然而,朴素的 chunking 方法很难准确地划分有意义的代码片段,导致边界定义问题以及包含无关或不完整的信息。我们发现,向 LLM 提供无效或不完整的代码片段实际上会损害性能并增加幻觉,而不是提供帮助。

Sweep AI 团队去年发布了一篇很棒的博文[译者注:原文链接已失效,此处为修正后的正确链接,指向 Sweep AI 团队在 GitHub 上关于代码分块方案的技术博客。],详细介绍了他们的代码 chunking 策略。他们开源了使用 concrete syntax tree (CST) parser 来创建连贯 chunk 的方法,该算法后来被 LlamaIndex 采用。

这是我们的起点,但我们在他们的方法中遇到了一些问题:

  1. 尽管有所改进,但 chunk 仍然不总是完整的,有时会丢失关键的 context,如 import 语句或类定义。

  2. 对可嵌入 chunk 大小的硬性限制并不总是允许捕获较大代码结构的完整 context。

  3. 该方法没有考虑到企业级代码库的独特挑战。

为了解决这些问题,我们开发了几种策略:

智能 Chunking 策略

Sweep AI 使用 static analysis(静态分析)实现了 chunking,这是对以前方法的巨大改进。但在当前节点超过 token 限制并开始将其子节点拆分为 chunk 而不考虑 context 的情况下,他们的方法并不是最优的。这可能导致在方法或 if 语句中间断开 chunk(例如,‘if’ 在一个 chunk 中,而 ‘else’ 在另一个中)。

为了缓解这个问题,我们使用特定于语言的 static analysis 将节点递归地划分为更小的 chunk,并执行追溯处理以重新添加任何被移除的关键 context。这使我们能够创建尊重代码结构的 chunk,将相关元素保持在一起。

from utilities import format_complex
class ComplexNumber:
def __init__(self, real, imag):
self.real = real
self.imag = imag
def modulus(self):
return math.sqrt(self.real**2 + self.imag**2)
def add(self, other):
return ComplexNumber(self.real + other.real, self.imag + other.imag)
def multiply(self, other):
new_real = self.real * other.real - self.imag * other.imag
new_imag = self.real * other.imag + self.imag * other.real
return ComplexNumber(new_real, new_imag)
def __str__(self):
return format_complex(self.real, self.imag)

Naive chunking:

def __str__(self):
return format_complex(self.real, self.imag)

[译者注:Naive chunking 只保留了方法本身,但丢失了其所属的类定义(ComplexNumber)、构造函数(init)以及依赖的 import 语句(format_complex),导致上下文不完整。]

Our chunking:

from utilities import format_complex

class ComplexNumber:
def __init__(self, real, imag):
self.real = real
self.imag = imag
# …
def __str__(self):
return format_complex(self.real, self.imag)

我们的 chunker 将关键 context 与类方法保持在一起,包括任何相关的 import 以及类定义和 init 方法,确保 AI 模型拥有理解和处理此代码所需的所有信息。

在 Chunk 中维护 Context

我们发现,embedding 较小的 chunk 通常会带来更好的性能。理想情况下,你希望拥有包含相关 context 的最小可能的 chunk——包含任何无关内容都会稀释 embedding 的语义含义。我们的目标是使 chunk 尽可能小,并将限制设定在 500 个字符左右。大型类或复杂的代码结构通常会超过此限制,导致代码表示不完整或碎片化。

因此,我们开发了一个系统,允许灵活的 chunk 大小,并确保关键 context(如类定义和 import 语句)包含在相关的 chunk 中。

对于一个大型类,我们可能会为单个方法分别创建 embedding 和索引,但在每个方法 chunk 中包含类定义和相关的 import。这样,当检索到特定方法时,AI 模型就拥有了理解和处理该方法所需的完整 context。

不同文件类型的特殊处理

不同的文件类型(例如代码文件、配置文件、文档)需要不同的 chunking 策略来维护其语义结构。

我们为各种文件类型实施了专门的 chunking 策略,特别关注像 OpenAPI/Swagger 规范这样具有复杂、互连结构的文件。

对于 OpenAPI 文件,我们不是按行或字符进行 chunking,而是按 endpoints 进行 chunking,确保每个 chunk 包含特定 API endpoint 的所有信息,包括其参数、响应和安全定义。

OpenAPI v3.0 – Naive Chunking

OpenAPI v3.0 – Intelligent Chunking

使用自然语言描述增强 Embeddings

代码 embeddings 通常无法捕捉代码的语义含义,特别是对于自然语言查询。

我们使用 LLM 为每个代码 chunk 生成自然语言描述。然后将这些描述与代码一起 embed,从而增强我们针对自然语言查询检索相关代码的能力。

对于前面展示的 map_finish_reason 函数:

# What is this?
## Helper utilities

def map_finish_reason( finish_reason: str,):
# openai supports 5 stop sequences - 'stop', 'length', 'function_call', 'content_filter', 'null'
# anthropic mapping
if finish_reason == "stop_sequence":
return "stop"
# cohere mapping - https://docs.cohere.com/reference/generate
elif finish_reason == "COMPLETE":
return "stop"
elif finish_reason == "MAX_TOKENS": # cohere + vertex ai
return "length"
elif finish_reason == "ERROR_TOXIC":
return "content_filter"
elif ( finish_reason == "ERROR" ): # openai currently doesn't support an 'error' finish reason
return "stop"
# huggingface mapping https://huggingface.github.io/text-generation-inference/#/Text%20Generation%20Inference/generate_stream
elif finish_reason == "eos_token" or finish_reason == "stop_sequence":
return "stop"
elif ( finish_reason == "FINISH_REASON_UNSPECIFIED"
or finish_reason == "STOP" ): # vertex ai - got from running `print(dir(response_obj.candidates[0].finish_reason))`: ['FINISH_REASON_UNSPECIFIED', 'MAX_TOKENS', 'OTHER', 'RECITATION', 'SAFETY', 'STOP',]
return "stop"
elif finish_reason == "SAFETY" or finish_reason == "RECITATION": # vertex ai
return "content_filter"
elif finish_reason == "STOP": # vertex ai
return "stop"
elif finish_reason == "end_turn" or finish_reason == "stop_sequence": # anthropic
return "stop"
elif finish_reason == "max_tokens": # anthropic
return "length"
elif finish_reason == "tool_use": # anthropic
return "tool_calls"
elif finish_reason == "content_filtered":
return "content_filter"
return finish_reason

我们可能会生成如下描述:

“Python function that standardizes finish reasons from various AI platforms, mapping platform-specific reasons to common terms like ‘stop’, ‘length’, and ‘content_filter’.”
(Python 函数,用于标准化来自各种 AI 平台的完成原因,将特定于平台的原因映射到通用术语,如 ‘stop’、‘length’ 和 ‘content_filter’。)

然后将此描述与代码一起 embed,从而改进对诸如“how to normalize AI completion statuses across different platforms”等查询的检索。这种方法旨在解决当前 embedding 模型中的差距,这些模型不是面向代码的,并且缺乏自然语言和代码之间的有效转换。

高级检索和排序

简单的向量相似度搜索通常会检索到不相关或脱离 context 的代码片段,特别是在拥有数百万索引 chunk 的大型多样化代码库中。

我们实施了两阶段检索过程。首先,我们从 vector store 中执行初始检索。然后,我们使用 LLM 根据结果与特定任务或查询的相关性对结果进行过滤和排序。

如果开发者查询“how to handle API rate limiting”,我们的系统可能会首先检索几个与 API 调用和错误处理相关的代码片段。然后,LLM 会在查询的 context 中分析这些片段,将那些专门处理速率限制逻辑的片段排在前面,并丢弃不相关的结果。

为企业仓库扩展 RAG

随着仓库数量增长到数千个,如果在每次查询时都跨所有仓库进行搜索,检索会变得嘈杂且效率低下。

我们正在开发 repo 级别的过滤策略,以便在深入研究单个代码 chunk 之前缩小搜索空间。这包括“golden repos”的概念——允许组织指定符合最佳实践并包含组织良好代码的特定仓库。

对于关于特定 microservice(微服务)架构模式的查询,我们的系统可能会首先根据 metadata 和高级内容分析识别出最有可能包含相关信息的 5-10 个仓库。然后,它会在这些仓库中执行详细的代码搜索,从而显著减少噪音并提高相关性。

RAG 基准测试和评估

由于缺乏标准化的 benchmarks,评估代码 RAG 系统的性能具有挑战性。

我们开发了一种多方面的评估方法,结合了自动化指标和来自企业客户的真实使用数据。

我们结合使用相关性评分(开发者实际使用检索到的代码片段的频率)、准确性指标(针对代码补全任务)和效率测量(响应时间、资源使用)。我们还与企业客户密切合作,收集反馈和真实的性能数据。

结论

为海量企业代码库实施 RAG 带来了超出典型 RAG 应用的独特挑战。通过专注于智能 chunking、增强的 embeddings、高级检索技术和可扩展架构,我们开发了一个能够有效导航和利用企业级代码库中蕴含的巨大知识的系统。

随着我们继续完善我们的方法,我们对 RAG 彻底改变开发者与大型复杂代码库交互方式的潜力感到兴奋。我们相信,这些技术不仅会提高生产力,还会提高大型组织内的代码质量和一致性。