百度文心一言文心千帆大模型 ERNIE-Bot-turbo调用示例（golang版本）

package lib

import (
    "bufio"
    "bytes"
    "encoding/json"
    "errors"
    "fmt"
    "github.com/tidwall/gjson"
    "io/ioutil"
    "net/http"
    "strings"
)

//百度文心一言ernie-bot-turbo
type erniebotturbo struct {
    appid       string
    apikey      string
    secretkey   string
    accesstoken string
}

func newerniebotturbo(appid, apikey, secretkey string) (*erniebotturbo, error) {

    m := &erniebotturbo{
        appid:     appid,
        apikey:    apikey,
        secretkey: secretkey,
    }
    var err error
    m.accesstoken, err = m.generateaccesstoken()
    if err != nil {
        return m, err
    }
    return m, nil
}

//获取access_token
func (this *erniebotturbo) generateaccesstoken() (string, error) {
    url := fmt.sprintf("https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=%s&client_secret=%s",
        this.apikey,
        this.secretkey)
    // 创建post请求
    req, err := http.newrequest("get", url, nil)
    if err != nil {
        fmt.println("创建请求失败：", err)
        return "", err
    }
    // 发送请求
    client := http.client{}
    response, err := client.do(req)
    if err != nil {
        fmt.println("发送请求失败：", err)
        return "", err
    }
    defer response.body.close()

    // 读取响应
    responsebody, err := ioutil.readall(response.body)
    if err != nil {
        return "", err
    }
    accesstoken := gjson.get(string(responsebody), "access_token").string()
    if accesstoken == "" {
        return "", errors.new("获取access_token失败")
    }
    this.accesstoken = accesstoken
    return accesstoken, nil
}

//流式请求接口
func (this *erniebotturbo) streamchat(messages []map[string]string) (*bufio.reader, error) {
    url := fmt.sprintf("https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/eb-instant?access_token=" + this.accesstoken)

    // 构建请求参数
    params := map[string]interface{}{
        "messages": messages,
        "stream":   true,
    }

    // 创建http请求的body
    jsonparams, err := json.marshal(params)
    requestbody := bytes.newbuffer(jsonparams)

    // 创建post请求
    req, err := http.newrequest("post", url, requestbody)
    if err != nil {
        fmt.println("创建请求失败：", err)
        return nil, err
    }
    // 设置请求头
    //req.header.set("access", "text/event-stream")
    // 发送请求
    client := http.client{}
    response, err := client.do(req)
    if err != nil {
        fmt.println("发送请求失败：", err)
        return nil, err
    }

    //defer response.body.close()

    // 读取响应
    // 读取响应体数据
    reader := bufio.newreader(response.body)
    return reader, nil
}
func (this *erniebotturbo) streamrecv(reader *bufio.reader) (string, error) {
waitfordata:
    line, err := reader.readstring('\n')
    if err != nil {
        return "", err
    }
    // 处理每行数据
    line = strings.trimspace(line)
    if line == "" {
        goto waitfordata
    }

    // 根据冒号分割每行数据的键值对
    parts := strings.splitn(line, ":", 2)
    if len(parts) != 2 {
        return "", errors.new("数据格式错误")
    }

    key := strings.trimspace(parts[0])
    value := strings.trimspace(parts[1])

    // 根据键的不同处理不同的字段
    switch key {
    case "data":
        // 设置event的数据
        return value, nil
        //case "meta":
        //    // 解析json格式的元数据
        //    return value, nil
    }
    goto waitfordata
    //return "", errors.new("finish")
}

//流式请求接口
func (this *erniebotturbo) embedding(input []string) (string, error) {
    url := fmt.sprintf("https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/embeddings/embedding-v1?access_token=" + this.accesstoken)

    // 构建请求参数
    params := map[string]interface{}{
        "input": input,
    }

    // 创建http请求的body
    jsonparams, err := json.marshal(params)
    requestbody := bytes.newbuffer(jsonparams)

    // 创建post请求
    req, err := http.newrequest("post", url, requestbody)
    if err != nil {
        fmt.println("创建请求失败：", err)
        return "", err
    }
    // 设置请求头
    //req.header.set("access", "text/event-stream")
    // 发送请求
    client := http.client{}
    response, err := client.do(req)
    if err != nil {
        fmt.println("发送请求失败：", err)
        return "", err
    }

    defer response.body.close()

    // 读取响应
    responsebody, err := ioutil.readall(response.body)
    if err != nil {
        return "", err
    }
    return string(responsebody), nil
}

package lib

import (
    "log"
    "testing"
)

//测试获取access_token
func testerniebotturbogenerateaccesstoken(t *testing.t) {
    appid := "35662533"
    apikey := "iq1ffkoqigtmtzqrfxorvq6t"
    secretkey := "qbzsofausl8ugt1gkgsdsjentqsjrotc"
    m, err := newerniebotturbo(appid, apikey, secretkey)
    log.println(m.accesstoken, err)
}

//测试流式输出
func testerniebotturbostreamchat(t *testing.t) {
    appid := "xxxx"
    apikey := "xxxxxxx"
    secretkey := "xxxxxxxxxjrotc"
    m, _ := newerniebotturbo(appid, apikey, secretkey)
    system := "假设你是一个程序员，你是程序员老狼的数字分身，你擅长计算机互联网程序开发，你的微信：llike620，官网：gofly.v1kf.com。你开发了唯一客服系统，你了解唯一客服系统的各种问题，你的回答应该是简洁的、不要超过50字。你只能根据知识信息回答问题。你绝对不能暴露自己是个“人工智能、大模型、ai、机器人、gpt”。如果用户问题里有自己联系方式，请回复“稍等，我们马上联系您”。请检测问题语种，并用该语种回答问题."
    prompt := []map[string]string{{"role": "user", "content": system + "我的问题是：你是谁"}}
    res, _ := m.streamchat(prompt)
    for {
        str, err := m.streamrecv(res)
        if err != nil {
            log.println(err)
            break
        }
        log.println(str, err)
    }
}

//测试embedding
func testerniebotturboembedding(t *testing.t) {
    appid := "xxxxxxx"
    apikey := "xxxxxxxxxxq6t"
    secretkey := "qxxxxxxxsofaxxxxxxxxxxxxxxxxxxxc"
    m, _ := newerniebotturbo(appid, apikey, secretkey)
    prompt := []string{"我的问题是：你是谁"}
    res, err := m.embedding(prompt)
    log.println(res, err)
}