Skip to content

Commit 699f9e6

Browse files
author
Craig D Wilhite
authored
Run App Studio project chat through Eino (#303)
* feat(app-studio): run project chat through Eino * chore(ci): trigger rerun for PR 303
1 parent c75e3a7 commit 699f9e6

5 files changed

Lines changed: 218 additions & 68 deletions

File tree

‎providers/app-studio/api/assistant_eino_engine.go‎

Lines changed: 150 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -21,23 +21,60 @@ import (
2121
"errors"
2222

2323
"github.com/cloudwego/eino/adk"
24+
"github.com/cloudwego/eino/schema"
2425
)
2526

2627
type projectEinoAssistantEngine struct {
27-
body projectAssistantEngine
28-
runner *adk.Runner
28+
body projectEinoAssistantBody
29+
newRunner projectEinoAssistantRunnerFactory
2930
}
3031

31-
// NewEinoAssistantEngine returns an Eino-backed assistant engine construction
32-
// proof. Later stack slices route normal assistant execution through this
33-
// runtime.
32+
type projectEinoAssistantBody func(
33+
context.Context,
34+
projectAssistantRunRequest,
35+
projectAssistantEventSink,
36+
) (projectAssistantRunResult, error)
37+
38+
type projectEinoAssistantRunner interface {
39+
Run(context.Context, []adk.Message, ...adk.AgentRunOption) *adk.AsyncIterator[*adk.AgentEvent]
40+
}
41+
42+
type projectEinoAssistantRunnerFactory func(context.Context, adk.Agent) projectEinoAssistantRunner
43+
44+
// NewEinoAssistantEngine returns the Eino-backed assistant engine. The App
45+
// Studio chat and tool loop runs as the body of an Eino ADK agent so execution
46+
// goes through Eino's runner pipeline.
3447
func NewEinoAssistantEngine(server *Server) projectAssistantEngine {
3548
return projectEinoAssistantEngine{
36-
body: projectChatCompletionAssistantEngine{server: server},
37-
runner: adk.NewRunner(context.Background(), adk.RunnerConfig{EnableStreaming: true}),
49+
body: newProjectEinoAssistantBody(server),
50+
newRunner: newProjectEinoAssistantRunner,
3851
}
3952
}
4053

54+
func newProjectEinoAssistantBody(server *Server) projectEinoAssistantBody {
55+
return func(ctx context.Context, req projectAssistantRunRequest, sink projectAssistantEventSink) (projectAssistantRunResult, error) {
56+
_ = sink
57+
if server == nil {
58+
return projectAssistantRunResult{}, errors.New("server is not configured")
59+
}
60+
if err := ctx.Err(); err != nil {
61+
return projectAssistantRunResult{}, err
62+
}
63+
reply, err := server.runProjectAssistantChatLoop(ctx, req)
64+
if err != nil {
65+
return projectAssistantRunResult{}, err
66+
}
67+
return projectAssistantRunResult{Content: reply}, nil
68+
}
69+
}
70+
71+
func newProjectEinoAssistantRunner(ctx context.Context, agent adk.Agent) projectEinoAssistantRunner {
72+
return adk.NewRunner(ctx, adk.RunnerConfig{
73+
Agent: agent,
74+
EnableStreaming: true,
75+
})
76+
}
77+
4178
func (e projectEinoAssistantEngine) StreamProjectAssistant(
4279
ctx context.Context,
4380
req projectAssistantRunRequest,
@@ -46,11 +83,113 @@ func (e projectEinoAssistantEngine) StreamProjectAssistant(
4683
if req.Project == nil {
4784
return projectAssistantRunResult{}, errors.New("project is required")
4885
}
49-
if e.runner == nil {
50-
return projectAssistantRunResult{}, errors.New("eino runner is not configured")
51-
}
5286
if e.body == nil {
5387
return projectAssistantRunResult{}, errors.New("assistant body is not configured")
5488
}
55-
return e.body.StreamProjectAssistant(ctx, req, sink)
89+
if e.newRunner == nil {
90+
return projectAssistantRunResult{}, errors.New("eino runner is not configured")
91+
}
92+
agent := projectEinoAssistantAgent{
93+
body: e.body,
94+
req: req,
95+
sink: sink,
96+
}
97+
runner := e.newRunner(ctx, agent)
98+
if runner == nil {
99+
return projectAssistantRunResult{}, errors.New("eino runner is not configured")
100+
}
101+
iter := runner.Run(ctx, []adk.Message{schema.UserMessage(projectEinoAssistantPrompt(req))})
102+
if iter == nil {
103+
return projectAssistantRunResult{}, errors.New("eino runner returned no event stream")
104+
}
105+
var result projectAssistantRunResult
106+
receivedOutput := false
107+
for {
108+
event, ok := iter.Next()
109+
if !ok {
110+
break
111+
}
112+
if event == nil {
113+
continue
114+
}
115+
if event.Err != nil {
116+
return projectAssistantRunResult{}, event.Err
117+
}
118+
if event.Output == nil {
119+
continue
120+
}
121+
if runResult, ok := event.Output.CustomizedOutput.(projectAssistantRunResult); ok {
122+
result = runResult
123+
receivedOutput = true
124+
continue
125+
}
126+
if event.Output.MessageOutput == nil {
127+
continue
128+
}
129+
msg, err := event.Output.MessageOutput.GetMessage()
130+
if err != nil {
131+
return projectAssistantRunResult{}, err
132+
}
133+
if msg != nil {
134+
result.Content = msg.Content
135+
receivedOutput = true
136+
}
137+
}
138+
if !receivedOutput {
139+
return projectAssistantRunResult{}, errors.New("eino runner completed without assistant output")
140+
}
141+
return result, nil
142+
}
143+
144+
type projectEinoAssistantAgent struct {
145+
body projectEinoAssistantBody
146+
req projectAssistantRunRequest
147+
sink projectAssistantEventSink
148+
}
149+
150+
func (a projectEinoAssistantAgent) Name(context.Context) string {
151+
return "app-studio-project-assistant"
152+
}
153+
154+
func (a projectEinoAssistantAgent) Description(context.Context) string {
155+
return "Runs App Studio project assistant turns."
156+
}
157+
158+
func (a projectEinoAssistantAgent) Run(
159+
ctx context.Context,
160+
input *adk.AgentInput,
161+
options ...adk.AgentRunOption,
162+
) *adk.AsyncIterator[*adk.AgentEvent] {
163+
_ = input
164+
_ = options
165+
iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
166+
go func() {
167+
defer gen.Close()
168+
if a.body == nil {
169+
gen.Send(&adk.AgentEvent{Err: errors.New("assistant body is not configured")})
170+
return
171+
}
172+
result, err := a.body(ctx, a.req, a.sink)
173+
if err != nil {
174+
gen.Send(&adk.AgentEvent{Err: err})
175+
return
176+
}
177+
gen.Send(&adk.AgentEvent{
178+
Output: &adk.AgentOutput{
179+
MessageOutput: &adk.MessageVariant{
180+
Message: schema.AssistantMessage(result.Content, nil),
181+
Role: schema.Assistant,
182+
},
183+
CustomizedOutput: result,
184+
},
185+
})
186+
}()
187+
return iter
188+
}
189+
190+
func projectEinoAssistantPrompt(req projectAssistantRunRequest) string {
191+
if req.Project != nil && req.Project.Name != "" {
192+
return "Run the App Studio project assistant for " + req.Project.Name + "."
193+
}
194+
return "Run the App Studio project assistant."
56195
}

‎providers/app-studio/api/assistant_eino_engine_test.go‎

Lines changed: 54 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@ import (
2121
"strings"
2222
"testing"
2323

24+
"github.com/cloudwego/eino/adk"
25+
2426
aiv1alpha1 "github.com/faroshq/provider-app-studio/apis/ai/v1alpha1"
2527
)
2628

@@ -38,12 +40,11 @@ func TestEinoAssistantEngineRequiresProject(t *testing.T) {
3840
}
3941
}
4042

41-
func TestEinoAssistantEngineRunsBody(t *testing.T) {
42-
engine, ok := NewEinoAssistantEngine(&Server{}).(projectEinoAssistantEngine)
43-
if !ok {
44-
t.Fatalf("engine = %T, want projectEinoAssistantEngine", NewEinoAssistantEngine(&Server{}))
43+
func TestEinoAssistantEngineRunsBodyThroughADKRunner(t *testing.T) {
44+
engine := projectEinoAssistantEngine{
45+
body: stubProjectAssistantBody(projectAssistantRunResult{Content: "from eino runner"}, nil),
46+
newRunner: newProjectEinoAssistantRunner,
4547
}
46-
engine.body = stubProjectAssistantEngine{result: projectAssistantRunResult{Content: "body result"}}
4748
result, err := engine.StreamProjectAssistant(
4849
context.Background(),
4950
projectAssistantRunRequest{
@@ -54,20 +55,58 @@ func TestEinoAssistantEngineRunsBody(t *testing.T) {
5455
if err != nil {
5556
t.Fatalf("StreamProjectAssistant returned error: %v", err)
5657
}
57-
if result.Content != "body result" {
58-
t.Fatalf("content = %q, want body result", result.Content)
58+
if result.Content != "from eino runner" {
59+
t.Fatalf("content = %q, want Eino runner result", result.Content)
5960
}
6061
}
6162

62-
type stubProjectAssistantEngine struct {
63-
result projectAssistantRunResult
64-
err error
63+
func TestEinoAssistantEngineRequiresRunnerOutput(t *testing.T) {
64+
engine := projectEinoAssistantEngine{
65+
body: stubProjectAssistantBody(projectAssistantRunResult{Content: "unused"}, nil),
66+
newRunner: func(context.Context, adk.Agent) projectEinoAssistantRunner {
67+
return emptyProjectEinoAssistantRunner{}
68+
},
69+
}
70+
_, err := engine.StreamProjectAssistant(
71+
context.Background(),
72+
projectAssistantRunRequest{
73+
Project: &aiv1alpha1.Project{},
74+
},
75+
nil,
76+
)
77+
if err == nil || !strings.Contains(err.Error(), "eino runner completed without assistant output") {
78+
t.Fatalf("StreamProjectAssistant error = %v, want missing runner output error", err)
79+
}
6580
}
6681

67-
func (e stubProjectAssistantEngine) StreamProjectAssistant(
82+
func TestServerRebuildsDefaultEinoAssistantEngine(t *testing.T) {
83+
server := &Server{}
84+
if _, ok := server.projectAssistantEngine().(projectEinoAssistantEngine); !ok {
85+
t.Fatalf("engine = %T, want projectEinoAssistantEngine", server.projectAssistantEngine())
86+
}
87+
}
88+
89+
func TestNewServerDefaultsToEinoAssistantEngine(t *testing.T) {
90+
server := NewWithWorkspace(nil, nil, nil, "", false)
91+
if _, ok := server.projectAssistantEngine().(projectEinoAssistantEngine); !ok {
92+
t.Fatalf("engine = %T, want projectEinoAssistantEngine", server.projectAssistantEngine())
93+
}
94+
}
95+
96+
func stubProjectAssistantBody(result projectAssistantRunResult, err error) projectEinoAssistantBody {
97+
return func(context.Context, projectAssistantRunRequest, projectAssistantEventSink) (projectAssistantRunResult, error) {
98+
return result, err
99+
}
100+
}
101+
102+
type emptyProjectEinoAssistantRunner struct{}
103+
104+
func (emptyProjectEinoAssistantRunner) Run(
68105
context.Context,
69-
projectAssistantRunRequest,
70-
projectAssistantEventSink,
71-
) (projectAssistantRunResult, error) {
72-
return e.result, e.err
106+
[]adk.Message,
107+
...adk.AgentRunOption,
108+
) *adk.AsyncIterator[*adk.AgentEvent] {
109+
iter, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]()
110+
gen.Close()
111+
return iter
73112
}

‎providers/app-studio/api/assistant_engine.go‎

Lines changed: 0 additions & 39 deletions
This file was deleted.

‎providers/app-studio/api/llm.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -323,7 +323,7 @@ func (s *Server) generateProjectAssistantStream(
323323
MCPInsecureSkipTLSVerify: s.mcpInsecureSkipTLSVerify,
324324
StreamCallbacks: callbacks,
325325
}
326-
result, err := projectChatCompletionAssistantEngine{server: s}.StreamProjectAssistant(ctx, req, nil)
326+
result, err := s.projectAssistantEngine().StreamProjectAssistant(ctx, req, nil)
327327
if err != nil {
328328
return "", err
329329
}

‎providers/app-studio/api/server.go‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,13 +38,15 @@ import (
3838
// Server holds the dependencies the project handlers need. clients builds a
3939
// per-(tenant, caller) dynamic client; store persists chat transcripts; hubBase
4040
// locates the hub's MCP virtual workspace; mcpInsecureSkipTLSVerify relaxes TLS
41-
// for dev MCP calls; workspaces stores project files owned by App Studio.
41+
// for dev MCP calls; workspaces stores project files owned by App Studio;
42+
// assistantEngine runs project assistant turns.
4243
type Server struct {
4344
clients *tenant.ClientFactory
4445
store store.Store
4546
workspaces *workspace.FileStore
4647
hubBase string
4748
mcpInsecureSkipTLSVerify bool
49+
assistantEngine projectAssistantEngine
4850
}
4951

5052
// New constructs a Server.
@@ -54,13 +56,22 @@ func New(clients *tenant.ClientFactory, msgStore store.Store, hubBase string, mc
5456

5557
// NewWithWorkspace constructs a Server with an explicit project workspace store.
5658
func NewWithWorkspace(clients *tenant.ClientFactory, msgStore store.Store, workspaces *workspace.FileStore, hubBase string, mcpInsecureSkipTLSVerify bool) *Server {
57-
return &Server{
59+
s := &Server{
5860
clients: clients,
5961
store: msgStore,
6062
workspaces: workspaces,
6163
hubBase: hubBase,
6264
mcpInsecureSkipTLSVerify: mcpInsecureSkipTLSVerify,
6365
}
66+
s.assistantEngine = NewEinoAssistantEngine(s)
67+
return s
68+
}
69+
70+
func (s *Server) projectAssistantEngine() projectAssistantEngine {
71+
if s.assistantEngine == nil {
72+
s.assistantEngine = NewEinoAssistantEngine(s)
73+
}
74+
return s.assistantEngine
6475
}
6576

6677
// Register mounts the project routes onto r. The hub backend proxy strips the

0 commit comments

Comments
 (0)