opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

init commit

Browse source
This commit is contained in:
Huaaudio 2026-03-21 05:52:21 +01:00
parent 100720bb74
commit 48c16f77eb
9 changed files with 402 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
go.mod
View file

@ -3,8 +3,8 @@ module github.com/sipeed/picoclaw
go 1.25.8
require (
github.com/BurntSushi/toml v1.6.0
fyne.io/systray v1.12.0
github.com/BurntSushi/toml v1.6.0
github.com/adhocore/gronx v1.19.6
github.com/anthropics/anthropic-sdk-go v1.26.0
github.com/bwmarrin/discordgo v0.29.0
@ -22,6 +22,8 @@ require (
github.com/mymmrac/telego v1.7.0
github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1
github.com/openai/openai-go/v3 v3.22.0
github.com/pion/rtp v1.8.7
github.com/pion/webrtc/v3 v3.3.6
github.com/rivo/tview v0.42.0
github.com/rs/zerolog v1.34.0
github.com/slack-go/slack v0.17.3
@ -53,6 +55,7 @@ require (
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/petermattis/goid v0.0.0-20260226131333-17d1149c6ac6 // indirect
github.com/pion/randutil v0.1.0 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
github.com/rivo/uniseg v0.4.7 // indirect

6
go.sum
View file

@ -162,6 +162,12 @@ github.com/openai/openai-go/v3 v3.22.0 h1:6MEoNoV8sbjOVmXdvhmuX3BjVbVdcExbVyGixi
github.com/openai/openai-go/v3 v3.22.0/go.mod h1:cdufnVK14cWcT9qA1rRtrXx4FTRsgbDPW7Ia7SS5cZo=
github.com/petermattis/goid v0.0.0-20260226131333-17d1149c6ac6 h1:rh2lKw/P/EqHa724vYH2+VVQ1YnW4u6EOXl0PMAovZE=
github.com/petermattis/goid v0.0.0-20260226131333-17d1149c6ac6/go.mod h1:pxMtw7cyUw6B2bRH0ZBANSPg+AoSud1I1iyJHI69jH4=
github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
github.com/pion/rtp v1.8.7 h1:qslKkG8qxvQ7hqaxkmL7Pl0XcUm+/Er7nMnu6Vq+ZxM=
github.com/pion/rtp v1.8.7/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU=
github.com/pion/webrtc/v3 v3.3.6 h1:7XAh4RPtlY1Vul6/GmZrv7z+NnxKA6If0KStXBI2ZLE=
github.com/pion/webrtc/v3 v3.3.6/go.mod h1:zyN7th4mZpV27eXybfR/cnUf3J2DRy8zw/mdjD9JTNM=
github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

28
pkg/bus/bus.go
View file

@ -34,6 +34,8 @@ type MessageBus struct {
inbound chan InboundMessage
outbound chan OutboundMessage
outboundMedia chan OutboundMediaMessage
audioChunks chan AudioChunk
voiceControls chan VoiceControl
closeOnce sync.Once
done chan struct{}
@ -47,6 +49,8 @@ func NewMessageBus() *MessageBus {
inbound: make(chan InboundMessage, defaultBusBufferSize),
outbound: make(chan OutboundMessage, defaultBusBufferSize),
outboundMedia: make(chan OutboundMediaMessage, defaultBusBufferSize),
audioChunks: make(chan AudioChunk, defaultBusBufferSize*4), // Audio chunks need more buffer
voiceControls: make(chan VoiceControl, defaultBusBufferSize),
done: make(chan struct{}),
}
}
@ -103,6 +107,22 @@ func (mb *MessageBus) OutboundMediaChan() <-chan OutboundMediaMessage {
return mb.outboundMedia
}
func (mb *MessageBus) PublishAudioChunk(ctx context.Context, chunk AudioChunk) error {
return publish(ctx, mb, mb.audioChunks, chunk)
}
func (mb *MessageBus) AudioChunksChan() <-chan AudioChunk {
return mb.audioChunks
}
func (mb *MessageBus) PublishVoiceControl(ctx context.Context, ctrl VoiceControl) error {
return publish(ctx, mb, mb.voiceControls, ctrl)
}
func (mb *MessageBus) VoiceControlsChan() <-chan VoiceControl {
return mb.voiceControls
}
// SetStreamDelegate registers a StreamDelegate (typically the channel Manager).
func (mb *MessageBus) SetStreamDelegate(d StreamDelegate) {
mb.streamDelegate.Store(d)
@ -132,6 +152,8 @@ func (mb *MessageBus) Close() {
close(mb.inbound)
close(mb.outbound)
close(mb.outboundMedia)
close(mb.audioChunks)
close(mb.voiceControls)
// clean up any remaining messages in channels
drained := 0
@ -144,6 +166,12 @@ func (mb *MessageBus) Close() {
for range mb.outboundMedia {
drained++
}
for range mb.audioChunks {
drained++
}
for range mb.voiceControls {
drained++
}
if drained > 0 {
logger.DebugCF("bus", "Drained buffered messages during close", map[string]any{

20
pkg/bus/types.go
View file

@ -51,3 +51,23 @@ type OutboundMediaMessage struct {
ChatID string `json:"chat_id"`
Parts []MediaPart `json:"parts"`
}
// AudioChunk represents a chunk of streaming voice data.
type AudioChunk struct {
SessionID string `json:"session_id"`
SpeakerID string `json:"speaker_id"` // User ID or SSRC
ChatID string `json:"chat_id"` // Where to respond
Sequence uint64 `json:"sequence"`
Timestamp uint32 `json:"timestamp"`
SampleRate int `json:"sample_rate"`
Channels int `json:"channels"`
Format string `json:"format"` // "opus", "pcm", etc
Data []byte `json:"data"`
}
// VoiceControl represents state or commands for voice sessions.
type VoiceControl struct {
SessionID string `json:"session_id"`
Type string `json:"type"` // "state", "command"
Action string `json:"action"` // "idle", "listening", "start", "stop"
}

6
pkg/channels/discord/discord.go
View file

@ -42,6 +42,7 @@ type DiscordChannel struct {
typingMu sync.Mutex
typingStop map[string]chan struct{} // chatID → stop signal
botUserID string // stored for mention checking
bus *bus.MessageBus
}
func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordChannel, error) {
@ -73,6 +74,7 @@ func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordC
config: cfg,
ctx: context.Background(),
typingStop: make(map[string]chan struct{}),
bus: bus,
}, nil
}
@ -321,6 +323,10 @@ func (c *DiscordChannel) handleMessage(s *discordgo.Session, m *discordgo.Messag
return
}
if c.handleVoiceCommand(s, m) {
return
}
// Check allowlist first to avoid downloading attachments for rejected users
sender := bus.SenderInfo{
Platform: "discord",

86
pkg/channels/discord/voice.go Normal file
View file

@ -0,0 +1,86 @@
package discord
import (
"fmt"
"github.com/bwmarrin/discordgo"
"github.com/sipeed/picoclaw/pkg/bus"
"github.com/sipeed/picoclaw/pkg/logger"
)
func (c *DiscordChannel) handleVoiceCommand(s *discordgo.Session, m *discordgo.MessageCreate) bool {
if m.Content == "!vc join" {
vs, err := s.State.VoiceState(m.GuildID, m.Author.ID)
if err != nil || vs == nil {
s.ChannelMessageSend(m.ChannelID, "You need to be in a voice channel first!")
return true
}
logger.InfoCF("discord", "Joining voice channel", map[string]any{"channel": vs.ChannelID})
vc, err := s.ChannelVoiceJoin(m.GuildID, vs.ChannelID, false, false)
if err != nil {
s.ChannelMessageSend(m.ChannelID, fmt.Sprintf("Failed to join voice channel: %v", err))
return true
}
go c.receiveVoice(vc, m.GuildID, m.ChannelID)
s.ChannelMessageSend(m.ChannelID, "Joined Voice Channel! Listening for audio...")
return true
} else if m.Content == "!vc leave" {
vc, exists := s.VoiceConnections[m.GuildID]
if exists && vc != nil {
vc.Disconnect()
s.ChannelMessageSend(m.ChannelID, "Left Voice Channel.")
} else {
s.ChannelMessageSend(m.ChannelID, "Not in a voice channel.")
}
return true
}
return false
}
func (c *DiscordChannel) receiveVoice(vc *discordgo.VoiceConnection, guildID string, chatID string) {
logger.InfoCF("discord", "Started listening for voice", map[string]any{"guild": guildID})
sessionID := fmt.Sprintf("discord_vc_%s", guildID)
c.bus.PublishVoiceControl(c.ctx, bus.VoiceControl{
SessionID: sessionID,
Type: "state",
Action: "listening",
})
var sequence uint64 = 0
for {
select {
case <-c.ctx.Done():
return
case p, ok := <-vc.OpusRecv:
if !ok {
logger.InfoCF("discord", "Voice channel closed", map[string]any{"guild": guildID})
return
}
if p == nil {
continue
}
sequence++
chunk := bus.AudioChunk{
SessionID: sessionID,
SpeakerID: fmt.Sprintf("%d", p.SSRC),
ChatID: chatID,
Sequence: sequence,
Timestamp: p.Timestamp,
SampleRate: 48000,
Channels: 2,
Format: "opus",
Data: p.Opus,
}
c.bus.PublishAudioChunk(c.ctx, chunk)
}
}
}

16
pkg/gateway/gateway.go
View file

@ -55,6 +55,7 @@ type services struct {
ChannelManager *channels.Manager
DeviceService *devices.Service
HealthServer *health.Server
VoiceAgentCancel context.CancelFunc
manualReloadChan chan struct{}
reloading atomic.Bool
}
@ -286,6 +287,12 @@ func setupAndStartServices(
if transcriber := voice.DetectTranscriber(cfg); transcriber != nil {
agentLoop.SetTranscriber(transcriber)
logger.InfoCF("voice", "Transcription enabled (agent-level)", map[string]any{"provider": transcriber.Name()})
// Start Voice Agent Orchestrator
vaCtx, vaCancel := context.WithCancel(context.Background())
runningServices.VoiceAgentCancel = vaCancel
voiceAgent := voice.NewAgent(msgBus, transcriber)
voiceAgent.Start(vaCtx)
}
enabledChannels := runningServices.ChannelManager.GetEnabledChannels()
@ -332,6 +339,9 @@ func stopAndCleanupServices(runningServices *services, shutdownTimeout time.Dura
if !isReload && runningServices.ChannelManager != nil {
runningServices.ChannelManager.StopAll(shutdownCtx)
}
if runningServices.VoiceAgentCancel != nil {
runningServices.VoiceAgentCancel()
}
if runningServices.DeviceService != nil {
runningServices.DeviceService.Stop()
}
@ -516,6 +526,12 @@ func restartServices(
al.SetTranscriber(transcriber)
if transcriber != nil {
logger.InfoCF("voice", "Transcription re-enabled (agent-level)", map[string]any{"provider": transcriber.Name()})
// Start Voice Agent Orchestrator on reload
vaCtx, vaCancel := context.WithCancel(context.Background())
runningServices.VoiceAgentCancel = vaCancel
voiceAgent := voice.NewAgent(msgBus, transcriber)
voiceAgent.Start(vaCtx)
} else {
logger.InfoCF("voice", "Transcription disabled", nil)
}

195
pkg/voice/agent.go Normal file
View file

@ -0,0 +1,195 @@
package voice
import (
"context"
"fmt"
"os"
"path/filepath"
"sync"
"time"
"github.com/pion/rtp"
"github.com/pion/webrtc/v3/pkg/media/oggwriter"
"github.com/sipeed/picoclaw/pkg/bus"
"github.com/sipeed/picoclaw/pkg/logger"
)
type speechAccumulator struct {
writer *oggwriter.OggWriter
file string
lastAudioAt time.Time
mu sync.Mutex
closed bool
chatID string
speakerID string
}
func (a *speechAccumulator) Push(chunk bus.AudioChunk) {
a.mu.Lock()
defer a.mu.Unlock()
if a.closed {
return
}
a.lastAudioAt = time.Now()
pkt := &rtp.Packet{
Header: rtp.Header{
SequenceNumber: uint16(chunk.Sequence),
Timestamp: chunk.Timestamp,
SSRC: uint32(chunk.Sequence), // Arbitrary dummy
},
Payload: chunk.Data,
}
if err := a.writer.WriteRTP(pkt); err != nil {
logger.ErrorCF("voice-agent", "Failed to write RTP", map[string]any{"error": err})
}
}
func (a *speechAccumulator) Close() {
a.mu.Lock()
defer a.mu.Unlock()
if !a.closed {
a.writer.Close()
a.closed = true
}
}
type Agent struct {
bus *bus.MessageBus
transcriber Transcriber
mu sync.Mutex
sessions map[string]*speechAccumulator // keyed by sessionID_speakerID
}
func NewAgent(mb *bus.MessageBus, t Transcriber) *Agent {
return &Agent{
bus: mb,
transcriber: t,
sessions: make(map[string]*speechAccumulator),
}
}
func (a *Agent) Start(ctx context.Context) {
logger.InfoCF("voice-agent", "Started Voice Agent orchestrator", nil)
go a.listenChunks(ctx)
go a.vadTick(ctx)
}
func (a *Agent) listenChunks(ctx context.Context) {
chunks := a.bus.AudioChunksChan()
for {
select {
case <-ctx.Done():
return
case chunk := <-chunks:
a.handleChunk(chunk)
}
}
}
func (a *Agent) handleChunk(chunk bus.AudioChunk) {
a.mu.Lock()
defer a.mu.Unlock()
key := fmt.Sprintf("%s_%s", chunk.SessionID, chunk.SpeakerID)
acc, exists := a.sessions[key]
if !exists {
filename := filepath.Join(os.TempDir(), fmt.Sprintf("voice_%s_%d.ogg", key, time.Now().UnixNano()))
writer, err := oggwriter.New(filename, uint32(chunk.SampleRate), uint16(chunk.Channels))
if err != nil {
logger.ErrorCF("voice-agent", "Failed to create OggWriter", map[string]any{"error": err})
return
}
acc = &speechAccumulator{
writer: writer,
file: filename,
lastAudioAt: time.Now(),
chatID: chunk.ChatID,
speakerID: chunk.SpeakerID,
}
a.sessions[key] = acc
logger.DebugCF("voice-agent", "Started accumulating voice", map[string]any{"key": key, "file": filename})
}
acc.Push(chunk)
}
func (a *Agent) vadTick(ctx context.Context) {
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
a.checkSilence(ctx)
}
}
}
func (a *Agent) checkSilence(ctx context.Context) {
a.mu.Lock()
now := time.Now()
var finished []*speechAccumulator
for key, acc := range a.sessions {
acc.mu.Lock()
last := acc.lastAudioAt
acc.mu.Unlock()
if now.Sub(last) > 1500*time.Millisecond {
acc.Close()
delete(a.sessions, key)
finished = append(finished, acc)
}
}
a.mu.Unlock()
for _, acc := range finished {
go a.processUtterance(ctx, acc)
}
}
func (a *Agent) processUtterance(ctx context.Context, acc *speechAccumulator) {
defer os.Remove(acc.file)
logger.InfoCF("voice-agent", "User finished speaking, transcribing...", map[string]any{"file": acc.file})
if a.transcriber == nil {
logger.ErrorCF("voice-agent", "No STT configured!", nil)
return
}
res, err := a.transcriber.Transcribe(ctx, acc.file)
if err != nil {
logger.ErrorCF("voice-agent", "Transcription failed", map[string]any{"error": err})
return
}
if res.Text == "" {
logger.DebugCF("voice-agent", "Ignored empty transcription", map[string]any{"file": acc.file})
return
}
logger.InfoCF("voice-agent", "Transcription result", map[string]any{"text": res.Text, "duration": res.Duration})
channelType := "discord"
a.bus.PublishInbound(ctx, bus.InboundMessage{
Channel: channelType,
SenderID: acc.speakerID,
ChatID: acc.chatID,
Content: res.Text,
Peer: bus.Peer{Kind: "channel", ID: acc.chatID},
Metadata: map[string]string{
"is_voice": "true",
},
})
}

70
pkg/voice/transcriber.go
View file

@ -21,6 +21,7 @@ import (
type Transcriber interface {
Name() string
Transcribe(ctx context.Context, audioFilePath string) (*TranscriptionResponse, error)
TranscribeData(ctx context.Context, data []byte, filename string) (*TranscriptionResponse, error)
}
type GroqTranscriber struct {
@ -48,45 +49,24 @@ func NewGroqTranscriber(apiKey string) *GroqTranscriber {
}
}
func (t *GroqTranscriber) Transcribe(ctx context.Context, audioFilePath string) (*TranscriptionResponse, error) {
logger.InfoCF("voice", "Starting transcription", map[string]any{"audio_file": audioFilePath})
audioFile, err := os.Open(audioFilePath)
if err != nil {
logger.ErrorCF("voice", "Failed to open audio file", map[string]any{"path": audioFilePath, "error": err})
return nil, fmt.Errorf("failed to open audio file: %w", err)
}
defer audioFile.Close()
fileInfo, err := audioFile.Stat()
if err != nil {
logger.ErrorCF("voice", "Failed to get file info", map[string]any{"path": audioFilePath, "error": err})
return nil, fmt.Errorf("failed to get file info: %w", err)
}
logger.DebugCF("voice", "Audio file details", map[string]any{
"size_bytes": fileInfo.Size(),
"file_name": filepath.Base(audioFilePath),
})
func (t *GroqTranscriber) TranscribeData(ctx context.Context, data []byte, filename string) (*TranscriptionResponse, error) {
logger.InfoCF("voice", "Starting memory transcription", map[string]any{"filename": filename, "bytes": len(data)})
var requestBody bytes.Buffer
writer := multipart.NewWriter(&requestBody)
part, err := writer.CreateFormFile("file", filepath.Base(audioFilePath))
part, err := writer.CreateFormFile("file", filename)
if err != nil {
logger.ErrorCF("voice", "Failed to create form file", map[string]any{"error": err})
return nil, fmt.Errorf("failed to create form file: %w", err)
}
copied, err := io.Copy(part, audioFile)
if err != nil {
if _, err := io.Copy(part, bytes.NewReader(data)); err != nil {
logger.ErrorCF("voice", "Failed to copy file content", map[string]any{"error": err})
return nil, fmt.Errorf("failed to copy file content: %w", err)
}
logger.DebugCF("voice", "File copied to request", map[string]any{"bytes_copied": copied})
if err = writer.WriteField("model", "whisper-large-v3"); err != nil {
if err = writer.WriteField("model", "whisper-large-v3-turbo"); err != nil {
logger.ErrorCF("voice", "Failed to write model field", map[string]any{"error": err})
return nil, fmt.Errorf("failed to write model field: %w", err)
}
@ -101,20 +81,52 @@ func (t *GroqTranscriber) Transcribe(ctx context.Context, audioFilePath string)
return nil, fmt.Errorf("failed to close multipart writer: %w", err)
}
return t.doRequest(ctx, &requestBody, writer.FormDataContentType(), int64(len(data)))
}
func (t *GroqTranscriber) Transcribe(ctx context.Context, audioFilePath string) (*TranscriptionResponse, error) {
logger.InfoCF("voice", "Starting transcription", map[string]any{"audio_file": audioFilePath})
audioFile, err := os.Open(audioFilePath)
if err != nil {
return nil, fmt.Errorf("failed to open audio file: %w", err)
}
defer audioFile.Close()
fileInfo, err := audioFile.Stat()
if err != nil {
return nil, err
}
var requestBody bytes.Buffer
writer := multipart.NewWriter(&requestBody)
part, err := writer.CreateFormFile("file", filepath.Base(audioFilePath))
if err != nil {
return nil, err
}
io.Copy(part, audioFile)
writer.WriteField("model", "whisper-large-v3")
writer.WriteField("response_format", "json")
writer.Close()
return t.doRequest(ctx, &requestBody, writer.FormDataContentType(), fileInfo.Size())
}
func (t *GroqTranscriber) doRequest(ctx context.Context, requestBody *bytes.Buffer, contentType string, fileSize int64) (*TranscriptionResponse, error) {
url := t.apiBase + "/audio/transcriptions"
req, err := http.NewRequestWithContext(ctx, "POST", url, &requestBody)
req, err := http.NewRequestWithContext(ctx, "POST", url, requestBody)
if err != nil {
logger.ErrorCF("voice", "Failed to create request", map[string]any{"error": err})
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", writer.FormDataContentType())
req.Header.Set("Content-Type", contentType)
req.Header.Set("Authorization", "Bearer "+t.apiKey)
logger.DebugCF("voice", "Sending transcription request to Groq API", map[string]any{
"url": url,
"request_size_bytes": requestBody.Len(),
"file_size_bytes": fileInfo.Size(),
"file_size_bytes": fileSize,
})
resp, err := t.httpClient.Do(req)