tRPC-Agent-Go 详解 - 腾讯开源的企业级 Go 语言 AI Agent 框架

AI人工智能 1 天前 0 1

前言

tRPC-Agent-Go 是腾讯开源的强大 Go 语言 AI Agent 框架，专为构建企业级智能代理系统而设计。它提供了多 Agent 协作、持久化记忆、丰富工具集成、可观测性等核心功能，特别适合构建客户服务机器人、数据分析助手、DevOps 自动化等生产级应用。本文详细介绍 tRPC-Agent-Go 的核心特性、安装配置、使用方法和实际应用场景。

项目地址：https://github.com/trpc-group/trpc-agent-go

官方文档：https://trpc-group.github.io/trpc-agent-go/

Stars：975+（快速增长中）

一、项目简介

1.1 核心定位

tRPC-Agent-Go 是一个用 Go 编写的企业级 AI Agent 框架，由腾讯 tRPC 团队开发。与 LangChain（Python）、AutoGen（Python）等框架不同，tRPC-Agent-Go 专注于 Go 语言生态，提供高性能、类型安全、生产就绪的 Agent 开发体验。

1.2 核心优势

特性	说明	对比优势
高性能	Go 语言原生，并发优秀	比 Python 方案快 5-10 倍
类型安全	Go 静态类型系统	编译时检查，减少运行时错误
企业级	腾讯生产环境验证	经过大规模业务考验
多 Agent 协作	链式、并行、图工作流	支持复杂协作模式
可观测性	内置 Langfuse、追踪	生产监控完善
MCP 协议	Model Context Protocol	标准化模型上下文

1.3 适用场景

客户服务机器人：理解上下文，解决复杂查询
数据分析助手：查询数据库、生成报告、提供洞察
DevOps 自动化：智能部署、监控、事件响应
业务流程自动化：多步骤工作流，人工介入支持
研究与知识管理：RAG 驱动的文档分析和问答

二、快速开始

2.1 环境要求

# Go 版本要求
go version >= 1.21

# 检查 Go 版本
go version

# 如果版本过低，升级 Go
# 访问 https://go.dev/dl/ 下载最新版本

2.2 安装

# 创建新项目
mkdir my-agent && cd my-agent
go mod init my-agent

# 安装 tRPC-Agent-Go
go get trpc.group/trpc-go/trpc-agent-go

# 安装依赖
go get github.com/trpc-group/trpc-codec/json
go get go.uber.org/zap

2.3 配置 API Key

# 设置环境变量
export OPENAI_API_KEY="your-openai-api-key"
export ANTHROPIC_API_KEY="your-anthropic-api-key"

# 或者创建 .env 文件
cat > .env << EOF
OPENAI_API_KEY=sk-your-key-here
ANTHROPIC_API_KEY=sk-ant-your-key-here
EOF

三、基础使用示例

3.1 最简单的 Agent

package main

import (
    "context"
    "fmt"
    "log"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

func main() {
    ctx := context.Background()
    
    // 创建 LLM 模型
    model := llm.NewOpenAIModel("gpt-4", nil)
    
    // 创建 Agent
    agent := llmagent.New("assistant",
        llmagent.WithModel(model))
    
    // 创建 Runner
    r := runner.NewRunner("app", agent)
    
    // 运行 Agent
    events, err := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("你好，请介绍一下自己"))
    if err != nil {
        log.Fatal(err)
    }
    
    // 处理事件流
    for event := range events {
        if event.Type == "message" {
            fmt.Printf("Agent: %s
", event.Data)
        }
    }
}

3.2 带工具使用的 Agent

package main

import (
    "context"
    "fmt"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/function"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

// 定义计算器函数
func calculate(a, b float64, op string) (float64, error) {
    switch op {
    case "add":
        return a + b, nil
    case "subtract":
        return a - b, nil
    case "multiply":
        return a * b, nil
    case "divide":
        if b == 0 {
            return 0, fmt.Errorf("division by zero")
        }
        return a / b, nil
    default:
        return 0, fmt.Errorf("unknown operation: %s", op)
    }
}

func main() {
    ctx := context.Background()
    
    // 创建工具
    calcTool := function.NewFunctionTool(
        calculate,
        function.WithName("calculator"),
        function.WithDescription("执行数学运算"),
    )
    
    // 创建模型
    model := llm.NewOpenAIModel("gpt-4", nil)
    
    // 创建带工具的 Agent
    agent := llmagent.New("math-assistant",
        llmagent.WithModel(model),
        llmagent.WithTools([]tool.Tool{calcTool}),
    )
    
    // 创建 Runner
    r := runner.NewRunner("app", agent)
    
    // 运行
    events, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("请计算 123 * 456 + 789"))
    
    for event := range events {
        if event.Type == "message" {
            fmt.Printf("结果：%s
", event.Data)
        }
    }
}

3.3 多 Agent 协作（链式）

package main

import (
    "context"
    "fmt"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/chainagent"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

func main() {
    ctx := context.Background()
    
    // 创建子 Agent
    analyzer := llmagent.New("analyzer",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("你是一个数据分析师，负责分析数据并提供洞察。"),
    )
    
    processor := llmagent.New("processor",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("你是一个数据处理专家，负责清洗和转换数据。"),
    )
    
    reporter := llmagent.New("reporter",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("你是一个技术作家，负责撰写清晰的报告。"),
    )
    
    // 创建链式 Agent
    pipeline := chainagent.New("data-pipeline",
        chainagent.WithSubAgents([]agent.Agent{
            analyzer,
            processor,
            reporter,
        }),
    )
    
    // 创建 Runner
    r := runner.NewRunner("app", pipeline)
    
    // 运行
    events, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("分析销售数据并生成报告"))
    
    for event := range events {
        if event.Type == "message" {
            fmt.Printf("%s
", event.Data)
        }
    }
}

3.4 并行 Agent 执行

package main

import (
    "context"
    "fmt"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/parallelagent"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

func main() {
    ctx := context.Background()
    
    // 创建多个任务 Agent
    task1 := llmagent.New("research",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("研究市场趋势"),
    )
    
    task2 := llmagent.New("competitor",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("分析竞争对手"),
    )
    
    task3 := llmagent.New("customer",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("分析客户反馈"),
    )
    
    // 创建并行 Agent
    parallel := parallelagent.New("concurrent-analysis",
        parallelagent.WithSubAgents([]agent.Agent{
            task1, task2, task3,
        }),
    )
    
    // 创建 Runner
    r := runner.NewRunner("app", parallel)
    
    // 运行
    events, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("进行全面的市场分析"))
    
    for event := range events {
        if event.Type == "message" {
            fmt.Printf("%s
", event.Data)
        }
    }
}

四、高级功能

4.1 持久化记忆系统

package main

import (
    "context"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/memory"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

func main() {
    ctx := context.Background()
    
    // 创建记忆服务
    memorySvc := memorysvc.NewInMemoryService()
    
    // 创建模型
    model := llm.NewOpenAIModel("gpt-4", nil)
    
    // 创建带记忆工具的 Agent
    agent := llmagent.New("assistant",
        llmagent.WithModel(model),
        llmagent.WithTools(memorySvc.Tools()),
    )
    
    // 创建 Runner（记忆服务在 Runner 级别管理）
    r := runner.NewRunner("app", agent,
        runner.WithMemoryService(memorySvc),
    )
    
    // 第一次对话
    events1, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("我叫小明，喜欢编程"))
    
    // 消耗事件
    for range events1 {}
    
    // 第二次对话（Agent 会记住之前的信息）
    events2, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("我叫什么名字？我喜欢什么？"))
    
    for event := range events2 {
        if event.Type == "message" {
            println(event.Data)
            // 输出：你叫小明，喜欢编程
        }
    }
}

4.2 GraphAgent（图工作流）

package main

import (
    "context"
    
    "trpc.group/trpc-go/trpc-agent-go/graphagent"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

func main() {
    ctx := context.Background()
    
    // 创建图 Agent
    graph := graphagent.New("workflow",
        graphagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
    )
    
    // 添加节点
    graph.AddNode("analyze",
        graphagent.WithSystemPrompt("分析用户需求"),
    )
    
    graph.AddNode("simple_task",
        graphagent.WithSystemPrompt("处理简单任务"),
    )
    
    graph.AddNode("complex_task",
        graphagent.WithSystemPrompt("处理复杂任务"),
    )
    
    graph.AddNode("review",
        graphagent.WithSystemPrompt("审查结果"),
    )
    
    // 添加边（条件路由）
    graph.AddEdge("analyze", "simple_task",
        graphagent.WithCondition("复杂度 < 5"))
    
    graph.AddEdge("analyze", "complex_task",
        graphagent.WithCondition("复杂度 >= 5"))
    
    graph.AddEdge("simple_task", "review")
    graph.AddEdge("complex_task", "review")
    
    // 创建 Runner
    r := runner.NewRunner("app", graph)
    
    // 运行
    events, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("帮我完成这个任务"))
    
    for event := range events {
        if event.Type == "message" {
            println(event.Data)
        }
    }
}

4.3 Agent Skills（技能系统）

package main

import (
    "context"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
    "trpc.group/trpc-go/trpc-agent-go/skill"
    "trpc.group/trpc-go/trpc-agent-go/skilltool"
    "trpc.group/trpc-go/trpc-agent-go/localexec"
)

func main() {
    ctx := context.Background()
    
    // 创建技能仓库（从本地文件系统）
    repo, _ := skill.NewFSRepository("./skills")
    
    // 或者从 HTTP URL 加载（支持 .zip/.tar.gz）
    // repo, _ := skill.NewFSRepository("https://example.com/skills.zip")
    
    // 创建技能工具
    tools := []tool.Tool{
        skilltool.NewLoadTool(repo),   // 加载技能
        skilltool.NewRunTool(repo, localexec.New()), // 执行技能
    }
    
    // 创建 Agent
    agent := llmagent.New("skilled-assistant",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithTools(tools),
    )
    
    // 创建 Runner
    r := runner.NewRunner("app", agent)
    
    // 运行（Agent 会自动加载和执行技能）
    events, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("使用数据分析技能处理销售数据"))
    
    for event := range events {
        if event.Type == "message" {
            println(event.Data)
        }
    }
}

4.4 可观测性集成（Langfuse）

package main

import (
    "context"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/langfuse"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
    "go.opentelemetry.io/otel/attribute"
)

func main() {
    ctx := context.Background()
    
    // 启动 Langfuse 集成
    clean, _ := langfuse.Start(ctx)
    defer clean(ctx)
    
    // 创建 Agent
    agent := llmagent.New("assistant",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
    )
    
    // 创建 Runner
    r := runner.NewRunner("app", agent)
    
    // 运行（带 Langfuse 属性）
    events, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("你好"),
        agent.WithSpanAttributes(
            attribute.String("langfuse.user.id", "user-1"),
            attribute.String("langfuse.session.id", "session-1"),
            attribute.String("langfuse.project", "my-app"),
        ),
    )
    
    for event := range events {
        if event.Type == "message" {
            println(event.Data)
        }
    }
    
    // 在 Langfuse Dashboard 查看追踪
    // https://cloud.langfuse.com
}

4.5 MCP 协议支持

package main

import (
    "context"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/mcptool"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

func main() {
    ctx := context.Background()
    
    // 创建 MCP 连接
    serverConn := /* MCP 服务器连接 */
    
    // 创建 MCP 工具
    mcpTool := mcptool.New(serverConn)
    
    // 创建 Agent
    agent := llmagent.New("mcp-assistant",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithTools([]tool.Tool{mcpTool}),
    )
    
    // 创建 Runner
    r := runner.NewRunner("app", agent)
    
    // 运行
    events, _ := r.Run(ctx, "user-1", "session-1",
        model.NewUserMessage("使用 MCP 工具查询数据库"))
    
    for event := range events {
        if event.Type == "message" {
            println(event.Data)
        }
    }
}

五、实际应用场景

场景 1：客户服务机器人

package main

import (
    "context"
    
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/chainagent"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

func main() {
    ctx := context.Background()
    
    // 意图识别 Agent
    intentAgent := llmagent.New("intent-classifier",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("分类用户意图：技术支持、账单查询、产品咨询、投诉建议"),
    )
    
    // 技术支持 Agent
    supportAgent := llmagent.New("tech-support",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("提供技术支持，解决技术问题"),
    )
    
    // 账单查询 Agent
    billingAgent := llmagent.New("billing",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithSystemPrompt("处理账单查询和支付问题"),
    )
    
    // 创建链式工作流
    pipeline := chainagent.New("customer-service",
        chainagent.WithSubAgents([]agent.Agent{
            intentAgent,
            // 根据意图路由到对应 Agent
            // 实际应用中可使用 GraphAgent 实现条件路由
        }),
    )
    
    r := runner.NewRunner("app", pipeline)
    
    events, _ := r.Run(ctx, "customer-1", "session-1",
        model.NewUserMessage("我的账户无法登录，怎么办？"))
    
    for event := range events {
        if event.Type == "message" {
            println(event.Data)
        }
    }
}

场景 2：DevOps 自动化

package main

import (
    "context"
    
    "trpc.group/trpc-go/trpc-agent-go/function"
    "trpc.group/trpc-go/trpc-agent-go/llm"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

// 部署函数
func deployService(serviceName, environment string) (string, error) {
    // 实际部署逻辑
    return "部署成功", nil
}

// 监控检查函数
func checkHealth(serviceName string) (map[string]interface{}, error) {
    // 健康检查逻辑
    return map[string]interface{}{"status": "healthy"}, nil
}

func main() {
    ctx := context.Background()
    
    // 创建部署工具
    deployTool := function.NewFunctionTool(
        deployService,
        function.WithName("deploy"),
        function.WithDescription("部署服务到指定环境"),
    )
    
    // 创建监控工具
    healthTool := function.NewFunctionTool(
        checkHealth,
        function.WithName("health_check"),
        function.WithDescription("检查服务健康状态"),
    )
    
    // 创建 DevOps Agent
    devopsAgent := llmagent.New("devops-assistant",
        llmagent.WithModel(llm.NewOpenAIModel("gpt-4", nil)),
        llmagent.WithTools([]tool.Tool{deployTool, healthTool}),
        llmagent.WithSystemPrompt("你是 DevOps 助手，负责部署和监控服务"),
    )
    
    r := runner.NewRunner("app", devopsAgent)
    
    events, _ := r.Run(ctx, "devops-user", "session-1",
        model.NewUserMessage("将 user-service 部署到生产环境，然后检查健康状态"))
    
    for event := range events {
        if event.Type == "message" {
            println(event.Data)
        }
    }
}

六、性能对比

指标	tRPC-Agent-Go	LangChain	AutoGen
语言	Go	Python	Python
并发性能	优秀（原生 goroutine）	一般	一般
类型安全	编译时检查	运行时检查	运行时检查
内存占用	低	中	中
启动时间	快	慢	慢
企业级特性	腾讯生产验证	社区驱动	微软研究

七、最佳实践

7.1 错误处理

// 带错误处理的运行
events, err := r.Run(ctx, "user-1", "session-1", msg)
if err != nil {
    log.Printf("运行失败：%v", err)
    // 实现重试逻辑或降级策略
}

// 处理事件流中的错误
for event := range events {
    switch event.Type {
    case "message":
        fmt.Println(event.Data)
    case "error":
        log.Printf("Agent 错误：%s", event.Data)
    case "tool_call":
        fmt.Println("调用工具...")
    }
}

7.2 资源管理

// 使用 defer 清理资源
func runAgent() {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
    defer cancel()
    
    r := runner.NewRunner("app", agent)
    
    events, err := r.Run(ctx, "user-1", "session-1", msg)
    if err != nil {
        return
    }
    
    // 确保消耗所有事件
    for range events {}
}

总结

tRPC-Agent-Go 是一个强大的企业级 Go 语言 AI Agent 框架，核心优势包括：

高性能：Go 语言原生，并发性能优秀
类型安全：编译时检查，减少运行时错误
多 Agent 协作：链式、并行、图工作流支持
持久化记忆：跨会话上下文管理
丰富工具：函数工具、MCP 协议、技能系统
可观测性：Langfuse 集成、追踪、监控
生产就绪：腾讯生产环境验证

适合场景：

需要高性能和并发能力的场景
企业级生产应用
Go 语言技术栈团队
需要类型安全和编译时检查的项目
复杂多 Agent 协作场景

项目地址：https://github.com/trpc-group/trpc-agent-go 🚀

注：本文基于 tRPC-Agent-Go 最新版本编写，具体 API 可能随版本更新而变化，请以官方文档为准。

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

AI Agent Go 语言 tRPC-Agent-Go 企业级框架多 Agent 协作腾讯开源

前言