Skip to content

Commit 786c303

Browse files
authored
[feat] Add experiemental support for accessing collector prior to first streamed value in go (#2146)
We now support streaming out thinking tokens in go prior to streaming out parsed chunks. ``` $> cd engine/generators/languages/go/generated_tests/sample $> go test -run TestOnTick 2025-07-16T01:22:47.374 [BAML INFO] BAML (v0.201.0) loaded Tick: 1 chunks Accumulated: 0 -------------------------------- Tick: 2 chunks Accumulated: 0 -------------------------------- Tick: 3 chunks Accumulated: 0 -------------------------------- Tick: 4 chunks Accumulated: 1 -------------------------------- Tick: 5 chunks Accumulated: 36 -------------------------------- Tick: 6 chunks Accumulated: 62 -------------------------------- Tick: 7 chunks Accumulated: 103 -------------------------------- Tick: 8 chunks Accumulated: 123 -------------------------------- Tick: 9 chunks Accumulated: 142 -------------------------------- Tick: 10 chunks Accumulated: 157 -------------------------------- Tick: 11 chunks Accumulated: 180 -------------------------------- Tick: 12 chunks Accumulated: 211 -------------------------------- Tick: 13 chunks Accumulated: 240 -------------------------------- Tick: 14 chunks Accumulated: 260 -------------------------------- Tick: 15 chunks Accumulated: 279 -------------------------------- Tick: 16 chunks Accumulated: 295 -------------------------------- Tick: 17 chunks Accumulated: 312 -------------------------------- Tick: 18 chunks Accumulated: 354 -------------------------------- Tick: 19 chunks Accumulated: 387 -------------------------------- Tick: 20 chunks Accumulated: 427 -------------------------------- Tick: 21 chunks Accumulated: 453 -------------------------------- Tick: 22 chunks Accumulated: 454 -------------------------------- Tick: 23 chunks Accumulated: 454 -------------------------------- Tick: 24 chunks Accumulated: 454 -------------------------------- Tick: 25 chunks Accumulated: 454 -------------------------------- Tick: 26 chunks Accumulated: 454 -------------------------------- Tick: 27 chunks Accumulated: 454 -------------------------------- Tick: 28 chunks Accumulated: 454 -------------------------------- Tick: 29 chunks Accumulated: 454 -------------------------------- Tick: 30 chunks Accumulated: 454 -------------------------------- Tick: 31 chunks Accumulated: 454 -------------------------------- Tick: 32 chunks Accumulated: 454 -------------------------------- Tick: 33 chunks Accumulated: 454 -------------------------------- Tick: 34 chunks Accumulated: 454 -------------------------------- Tick: 35 chunks Accumulated: 454 -------------------------------- Tick: 36 chunks Accumulated: 454 -------------------------------- Tick: 37 chunks Accumulated: 454 -------------------------------- Tick: 38 chunks Accumulated: 454 -------------------------------- Tick: 39 chunks Accumulated: 454 -------------------------------- Tick: 40 chunks Accumulated: 454 -------------------------------- Tick: 41 chunks Accumulated: 454 -------------------------------- Tick: 42 chunks Accumulated: 454 -------------------------------- Tick: 43 chunks Accumulated: 454 -------------------------------- Tick: 44 chunks Accumulated: 454 -------------------------------- Tick: 45 chunks Accumulated: 454 -------------------------------- Tick: 46 chunks Accumulated: 454 -------------------------------- Tick: 47 chunks Accumulated: 454 -------------------------------- Tick: 48 chunks Accumulated: 454 -------------------------------- Tick: 49 chunks Accumulated: 454 -------------------------------- Tick: 50 chunks Accumulated: 454 -------------------------------- Tick: 51 chunks Accumulated: 454 -------------------------------- Tick: 52 chunks Accumulated: 454 -------------------------------- 2025-07-16T01:22:54.638 [BAML INFO] Function Foo: Client: Thinking (claude-sonnet-4-20250514) - 7237ms. StopReason: end_turn. Tokens(in/out): 145/358 ---PROMPT--- user: Fill out this data model with some examples. Answer in JSON using any of these schemas: { type: "example_2", item: { type: "example_1", a: int, b: string, }, element: string, element2: string, } or { type: "example_1", a: int, b: string, } use 8192 somewhere in the data model ---LLM REPLY--- Here are examples using both schemas: ```json { "type": "example_2", "item": { "type": "example_1", "a": 8192, "b": "buffer_size" }, "element": "memory_allocation", "element2": "high_performance_mode" } ``` ```json { "type": "example_1", "a": 8192, "b": "bytes_per_block" } ``` ```json { "type": "example_2", "item": { "type": "example_1", "a": 42, "b": "processing_threads" }, "element": "cache_size_8192", "element2": "optimization_enabled" } ``` ---Parsed Response (class Example2)--- { "type": "example_2", "item": { "type": "example_1", "a": 8192, "b": "buffer_size" }, "element": "memory_allocation", "element2": "high_performance_mode" } Tick: 1 chunks Accumulated: 0 -------------------------------- Tick: 2 chunks Accumulated: 0 -------------------------------- Tick: 3 chunks Accumulated: 0 -------------------------------- Tick: 4 chunks Accumulated: 1 -------------------------------- Tick: 5 chunks Accumulated: 36 -------------------------------- Tick: 6 chunks Accumulated: 84 -------------------------------- Tick: 7 chunks Accumulated: 127 -------------------------------- Tick: 8 chunks Accumulated: 160 -------------------------------- Tick: 9 chunks Accumulated: 200 -------------------------------- Tick: 10 chunks Accumulated: 244 -------------------------------- Tick: 11 chunks Accumulated: 275 -------------------------------- Tick: 12 chunks Accumulated: 308 -------------------------------- Tick: 13 chunks Accumulated: 342 -------------------------------- Tick: 14 chunks Accumulated: 400 -------------------------------- ^[[ATick: 15 chunks Accumulated: 438 -------------------------------- Tick: 16 chunks Accumulated: 464 -------------------------------- Tick: 17 chunks Accumulated: 487 -------------------------------- Tick: 18 chunks Accumulated: 527 -------------------------------- Tick: 19 chunks Accumulated: 557 -------------------------------- Tick: 20 chunks Accumulated: 585 -------------------------------- Tick: 21 chunks Accumulated: 585 -------------------------------- Tick: 22 chunks Accumulated: 585 -------------------------------- Tick: 23 chunks Accumulated: 585 -------------------------------- Tick: 24 chunks Accumulated: 585 -------------------------------- Tick: 25 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x140003b2de0} Tick: 26 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x14000424150} Tick: 27 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x1400035b5f0} Tick: 28 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x14000424540} Tick: 29 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x14000306360} Tick: 30 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x14000424510} Tick: 31 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x14000307f20} Tick: 32 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x1400021f2c0} Tick: 33 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x14000239080} Tick: 34 chunks Accumulated: 585 -------------------------------- Stream: &{variant:Example2 variant_Example:<nil> variant_Example2:0x14000244f00} 2025-07-16T01:23:01.230 [BAML INFO] Function Foo: Client: Thinking (claude-sonnet-4-20250514) - 6578ms. StopReason: end_turn. Tokens(in/out): 145/280 ---PROMPT--- user: Fill out this data model with some examples. Answer in JSON using any of these schemas: { type: "example_2", item: { type: "example_1", a: int, b: string, }, element: string, element2: string, } or { type: "example_1", a: int, b: string, } use 8192 somewhere in the data model ---LLM REPLY--- ```json { "type": "example_2", "item": { "type": "example_1", "a": 8192, "b": "buffer_size_configuration" }, "element": "memory_management", "element2": "system_optimization" } ``` ---Parsed Response (class Example2)--- { "type": "example_2", "item": { "type": "example_1", "a": 8192, "b": "buffer_size_configuration" }, "element": "memory_management", "element2": "system_optimization" } final &{Example2 <nil> 0x1400023d4a0} PASS ok sample 14.069s ```
1 parent de024cf commit 786c303

88 files changed

Lines changed: 9434 additions & 3105 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎engine/baml-runtime/src/cli/serve/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -448,6 +448,7 @@ Tip: test that the server is up using `curl http://localhost:{}/_debug/ping`
448448
Ok(mut result_stream) => {
449449
let (result, _trace_id) = result_stream
450450
.run(
451+
None::<fn()>,
451452
Some(move |result| {
452453
// If the receiver is closed (either because it called close or it was dropped),
453454
// we can't really do anything

‎engine/baml-runtime/src/internal/llm_client/orchestrator/stream.rs‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,13 @@ use crate::{
2424
FunctionResult, RuntimeContext,
2525
};
2626

27-
pub async fn orchestrate_stream<F>(
27+
pub async fn orchestrate_stream<F, G>(
2828
iter: OrchestratorNodeIterator,
2929
ir: &IntermediateRepr,
3030
ctx: &RuntimeContext,
3131
prompt: &PromptRenderer,
3232
params: &BamlValue,
33+
on_tick_fn: Option<G>,
3334
partial_parse_fn: impl Fn(&str) -> Result<ResponseBamlValue>,
3435
parse_fn: impl Fn(&str) -> Result<ResponseBamlValue>,
3536
on_event: Option<F>,
@@ -43,6 +44,7 @@ pub async fn orchestrate_stream<F>(
4344
)
4445
where
4546
F: Fn(FunctionResult),
47+
G: Fn(),
4648
{
4749
let mut results = Vec::new();
4850
let mut total_sleep_duration = std::time::Duration::from_secs(0);
@@ -67,6 +69,9 @@ where
6769
let final_response = match stream_res {
6870
Ok(response) => response
6971
.map(|stream_part| {
72+
if let Some(on_tick) = on_tick_fn.as_ref() {
73+
on_tick();
74+
}
7075
if let Some(on_event) = on_event.as_ref() {
7176
if let LLMResponse::Success(s) = &stream_part {
7277
let response_value = partial_parse_fn(&s.content);

‎engine/baml-runtime/src/lib.rs‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -514,7 +514,14 @@ impl BamlRuntime {
514514
vec![],
515515
)?;
516516
let (response_res, call_uuid) = stream
517-
.run(on_event, ctx, type_builder.as_ref(), None, env_vars.clone())
517+
.run(
518+
None::<fn()>,
519+
on_event,
520+
ctx,
521+
type_builder.as_ref(),
522+
None,
523+
env_vars.clone(),
524+
)
518525
.await;
519526
let res = response_res?;
520527
let (_, llm_resp, val) = res

‎engine/baml-runtime/src/tracingv2/storage/storage.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -248,7 +248,6 @@ fn build_function_log(
248248
let entry = calls_map.entry(rid.clone()).or_default();
249249

250250
// find or insert the event
251-
252251
match &mut entry.http_response_stream {
253252
Some(stream) => {
254253
stream.lock().unwrap().push(http_res_stream.clone());
@@ -370,7 +369,8 @@ fn build_function_log(
370369
};
371370

372371
let new_arc = Arc::new(Mutex::new(function_log_inner));
373-
{
372+
// only cache if we we've finished the function
373+
if function_end_time.is_some() {
374374
// Insert into the cache
375375
let mut lock = storage.function_inners.lock().unwrap();
376376
lock.insert(function_id.clone(), new_arc.clone());

‎engine/baml-runtime/src/types/stream.rs‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -63,8 +63,9 @@ first.scope.clone();
6363

6464
impl FunctionResultStream {
6565
#[cfg(not(target_arch = "wasm32"))]
66-
pub fn run_sync<F>(
66+
pub fn run_sync<F, G>(
6767
&mut self,
68+
on_tick: Option<G>,
6869
on_event: Option<F>,
6970
ctx: &RuntimeContextManager,
7071
tb: Option<&TypeBuilder>,
@@ -73,14 +74,16 @@ impl FunctionResultStream {
7374
) -> (Result<FunctionResult>, baml_ids::FunctionCallId)
7475
where
7576
F: Fn(FunctionResult),
77+
G: Fn(),
7678
{
7779
let rt = self.tokio_runtime.clone();
78-
let fut = self.run(on_event, ctx, tb, cb, env_vars);
80+
let fut = self.run(on_tick, on_event, ctx, tb, cb, env_vars);
7981
rt.block_on(fut)
8082
}
8183

82-
pub async fn run<F>(
84+
pub async fn run<F, G>(
8385
&mut self,
86+
on_tick: Option<G>,
8487
on_event: Option<F>,
8588
ctx: &RuntimeContextManager,
8689
tb: Option<&TypeBuilder>,
@@ -89,6 +92,7 @@ impl FunctionResultStream {
8992
) -> (Result<FunctionResult>, baml_ids::FunctionCallId)
9093
where
9194
F: Fn(FunctionResult),
95+
G: Fn(),
9296
{
9397
let mut local_orchestrator = Vec::new();
9498
std::mem::swap(&mut local_orchestrator, &mut self.orchestrator);
@@ -114,6 +118,7 @@ impl FunctionResultStream {
114118
&rctx,
115119
&self.renderer,
116120
&baml_types::BamlValue::Map(self.prepared_func.value.clone()),
121+
on_tick,
117122
|content| self.renderer.parse(self.ir.as_ref(), &rctx, content, true),
118123
|content| self.renderer.parse(self.ir.as_ref(), &rctx, content, false),
119124
on_event,

‎engine/generators/data/sample/baml_src/main.baml‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
class Example {
2+
type "example_1" @stream.not_null
23
a int
34
b string
45

@@ -7,13 +8,26 @@ class Example {
78
}
89

910
class Example2 {
11+
type "example_2" @stream.not_null
1012
item Example
1113
element string
1214
element2 string
1315
}
1416

17+
client<llm> Thinking {
18+
provider "anthropic"
19+
options {
20+
model "claude-sonnet-4-20250514"
21+
thinking {
22+
type enabled
23+
budget_tokens 1024
24+
}
25+
max_tokens 2048
26+
}
27+
}
28+
1529
function Foo(x: int) -> Example2 | Example {
16-
client "openai/gpt-4o-mini"
30+
client Thinking
1731
prompt #"
1832
Fill out this data model with some examples.
1933

‎engine/generators/data/sample/go/go.mod‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,15 @@ module sample
22

33
go 1.24.0
44

5-
require github.com/boundaryml/baml v0.90.0
5+
require (
6+
github.com/boundaryml/baml v0.90.0
7+
github.com/tidwall/gjson v1.18.0
8+
)
69

7-
require google.golang.org/protobuf v1.36.6 // indirect
10+
require (
11+
github.com/tidwall/match v1.1.1 // indirect
12+
github.com/tidwall/pretty v1.2.0 // indirect
13+
google.golang.org/protobuf v1.36.6 // indirect
14+
)
815

916
replace github.com/boundaryml/baml => ../../../../../../

‎engine/generators/data/sample/go/go.sum‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,12 @@ github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
66
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
77
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
88
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
9+
github.com/tidwall/gjson v1.18.0 h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY=
10+
github.com/tidwall/gjson v1.18.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
11+
github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
12+
github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
13+
github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
14+
github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
915
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
1016
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
1117
google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=

‎engine/generators/data/sample/go/main_test.go‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,106 @@ import (
1010
"testing"
1111

1212
baml "github.com/boundaryml/baml/engine/language_client_go/pkg"
13+
"github.com/tidwall/gjson"
1314
)
1415

16+
func getOnTick() (baml.TickCallback, *string, *int) {
17+
var lastThinking string
18+
var tickCount int
19+
onTick := func(ctx context.Context, reason baml.TickReason, log baml.FunctionLog) baml.FunctionSignal {
20+
tickCount++
21+
calls, err := log.Calls()
22+
if err != nil {
23+
fmt.Println("Error getting selected call: ", err)
24+
return nil
25+
}
26+
27+
if len(calls) == 0 {
28+
fmt.Println("No calls found")
29+
return nil
30+
}
31+
32+
selectedCall := calls[len(calls)-1]
33+
34+
if as_stream, ok := selectedCall.(baml.LLMStreamCall); ok {
35+
stream, err := as_stream.SSEChunks()
36+
if err != nil {
37+
fmt.Println("Error getting stream: ", err)
38+
return nil
39+
}
40+
fmt.Println("Tick: ", len(stream), " chunks")
41+
accumulated := ""
42+
for _, chunk := range stream {
43+
text, err := chunk.Text()
44+
if err != nil {
45+
// this is an openai stream
46+
} else {
47+
if content := gjson.Get(text, "delta.thinking"); content.Exists() {
48+
accumulated += content.String()
49+
}
50+
}
51+
}
52+
fmt.Println("Accumulated: ", accumulated)
53+
lastThinking = accumulated
54+
fmt.Println("--------------------------------")
55+
} else {
56+
fmt.Println("Response is not a stream: ", selectedCall)
57+
}
58+
59+
return nil
60+
}
61+
62+
return onTick, &lastThinking, &tickCount
63+
}
64+
65+
func TestOnTickRequest(t *testing.T) {
66+
ctx := context.Background()
67+
68+
onTick, lastThinking, tickCount := getOnTick()
69+
70+
_, err := b.Foo(ctx, 8192, b.WithExperimentalOnTick(onTick))
71+
if err != nil {
72+
t.Fatalf("Error in Foo: %v", err)
73+
}
74+
75+
if *tickCount < 10 {
76+
t.Errorf("Expected more than 10 ticks, got %d", *tickCount)
77+
}
78+
79+
if *lastThinking == "" {
80+
t.Errorf("Expected thinking, got %s", *lastThinking)
81+
}
82+
}
83+
84+
func TestOnTickStream(t *testing.T) {
85+
ctx := context.Background()
86+
87+
onTick, lastThinking, tickCount := getOnTick()
88+
89+
result, err := b.Stream.Foo(ctx, 8192, b.WithExperimentalOnTick(onTick))
90+
if err != nil {
91+
t.Fatalf("Error in Foo: %v", err)
92+
}
93+
for result := range result {
94+
if result.IsError {
95+
t.Fatalf("Error in Foo: %v", result.Error)
96+
} else if result.IsFinal {
97+
final := result.Final()
98+
fmt.Println("final", final)
99+
} else {
100+
fmt.Printf("Stream: %+v\n", result.Stream())
101+
}
102+
}
103+
104+
if *tickCount < 10 {
105+
t.Errorf("Expected more than 10 ticks, got %d", *tickCount)
106+
}
107+
108+
if *lastThinking == "" {
109+
t.Errorf("Expected thinking, got %s", *lastThinking)
110+
}
111+
}
112+
15113
func TestFoo(t *testing.T) {
16114
t.Parallel()
17115
ctx := context.Background()

0 commit comments

Comments
 (0)