Source file convert_stream.ml
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
open Melange_json.Primitives
type block_state =
| Text_block
| Tool_use_block of {
id : string;
name : string;
}
| Thinking_block
type content_block_info = {
type_ : string; [@json.key "type"]
id : string option; [@json.default None]
name : string option; [@json.default None]
}
[@@json.allow_extra_fields] [@@deriving of_json]
type content_block_start_event = {
index : int;
content_block : content_block_info;
}
[@@json.allow_extra_fields] [@@deriving of_json]
type delta_info = {
type_ : string; [@json.key "type"]
text : string option; [@json.default None]
partial_json : string option; [@json.default None]
thinking : string option; [@json.default None]
}
[@@json.allow_extra_fields] [@@deriving of_json]
type content_block_delta_event = {
index : int;
delta : delta_info;
}
[@@json.allow_extra_fields] [@@deriving of_json]
type content_block_stop_event = { index : int } [@@json.allow_extra_fields] [@@deriving of_json]
type message_delta_info = { stop_reason : string option [@json.default None] }
[@@json.allow_extra_fields] [@@deriving of_json]
type message_delta_event = {
delta : message_delta_info;
usage : Convert_usage.anthropic_usage option; [@json.default None]
}
[@@json.allow_extra_fields] [@@deriving of_json]
type message_start_message = { usage : Convert_usage.anthropic_usage option [@json.default None] }
[@@json.allow_extra_fields] [@@deriving of_json]
type message_start_event = { message : message_start_message } [@@json.allow_extra_fields] [@@deriving of_json]
type error_info = {
type_ : string; [@json.key "type"] [@json.default "unknown"]
message : string; [@json.default ""]
}
[@@json.allow_extra_fields] [@@deriving of_json]
type error_event = { error : error_info } [@@json.allow_extra_fields] [@@deriving of_json]
let transform events ~warnings =
let blocks : (int, block_state) Hashtbl.t = Hashtbl.create 8 in
let is_first = ref true in
let start_usage : Convert_usage.anthropic_usage option ref = ref None in
let stream, push = Lwt_stream.create () in
Lwt.async (fun () ->
let%lwt () =
Lwt_stream.iter
(fun (evt : Sse.event) ->
try
let json = Yojson.Basic.from_string evt.data in
match evt.event_type with
| "message_start" ->
(try
let evt = message_start_event_of_json json in
start_usage := evt.message.usage
with Melange_json.Of_json_error _ -> ());
if !is_first then begin
push (Some (Ai_provider.Stream_part.Stream_start { warnings }));
is_first := false
end
| "content_block_start" ->
let { index; content_block } = content_block_start_event_of_json json in
(match content_block.type_ with
| "text" -> Hashtbl.replace blocks index Text_block
| "tool_use" ->
(match content_block.id, content_block.name with
| Some id, Some name -> Hashtbl.replace blocks index (Tool_use_block { id; name })
| _ -> ())
| "thinking" -> Hashtbl.replace blocks index Thinking_block
| _ -> ())
| "content_block_delta" ->
let { index; delta } = content_block_delta_event_of_json json in
(match delta.type_ with
| "text_delta" ->
(match delta.text with
| Some text -> push (Some (Ai_provider.Stream_part.Text { text }))
| None -> ())
| "input_json_delta" ->
(match delta.partial_json with
| Some partial ->
(match Hashtbl.find_opt blocks index with
| Some (Tool_use_block { id; name }) ->
push
(Some
(Ai_provider.Stream_part.Tool_call_delta
{
tool_call_type = "function";
tool_call_id = id;
tool_name = name;
args_text_delta = partial;
}))
| _ -> ())
| None -> ())
| "thinking_delta" ->
(match delta.thinking with
| Some text -> push (Some (Ai_provider.Stream_part.Reasoning { text; signature = None }))
| None -> ())
| _ -> ())
| "content_block_stop" ->
let { index } = content_block_stop_event_of_json json in
(match Hashtbl.find_opt blocks index with
| Some (Tool_use_block { id; _ }) ->
push (Some (Ai_provider.Stream_part.Tool_call_finish { tool_call_id = id }))
| _ -> ());
Hashtbl.remove blocks index
| "message_delta" ->
let { delta; usage } = message_delta_event_of_json json in
let has_cache_signal (u : Convert_usage.anthropic_usage) =
Option.is_some u.cache_read_input_tokens
|| Option.is_some u.cache_creation_input_tokens
|| Option.is_some u.cache_creation
in
let anthropic_usage =
match usage, !start_usage with
| None, _ -> !start_usage
| Some u, _ when has_cache_signal u -> Some u
| Some u, None -> Some u
| Some u, Some s -> Some { s with output_tokens = u.output_tokens }
in
let usage_ai =
match anthropic_usage with
| Some u -> Convert_usage.to_usage u
| None -> { Ai_provider.Usage.input_tokens = 0; output_tokens = 0; total_tokens = None }
in
let provider_metadata =
match anthropic_usage with
| Some u when has_cache_signal u -> Some (Convert_usage.to_provider_metadata u)
| _ -> None
in
push
(Some
(Ai_provider.Stream_part.Finish
{
finish_reason = Convert_response.map_stop_reason delta.stop_reason;
usage = usage_ai;
provider_metadata;
}))
| "message_stop" | "ping" -> ()
| "error" ->
let error_type, message =
try
let { error = { type_; message } } = error_event_of_json json in
type_, message
with Melange_json.Of_json_error _ -> "unknown", evt.data
in
push
(Some
(Ai_provider.Stream_part.Error
{
error =
Ai_provider.Provider_error.make_api_error ~provider:"anthropic" ~status:0
~body:(Printf.sprintf "%s: %s" error_type message)
~is_retryable:false ();
}))
| _ -> ()
with (Yojson.Json_error _ | Melange_json.Of_json_error _) as exn ->
push
(Some
(Ai_provider.Stream_part.Error
{
error =
{
Ai_provider.Provider_error.provider = "anthropic";
kind = Deserialization_error { message = Printexc.to_string exn; raw = evt.data };
is_retryable = false;
};
})))
events
in
push None;
Lwt.return_unit);
stream