1use super::super::super::{
2 GenerationResources, GenerationStateAccess,
3 chat_responses::{ChatViaResponses, ResponsesViaChat},
4};
5use super::super::{StreamInvocation, StreamSettings, StreamTarget};
6use super::RequestMode;
7use crate::{
8 capability::{ResourceAccess, StateStore},
9 transform::{TransformError, generate::chat_responses::stream as p},
10 wire::{
11 DeclaredFields,
12 openai::{chat as h, responses as r},
13 },
14};
15
16impl ChatViaResponses {
17 pub async fn prepare_stream<S: StateStore>(
18 input: h::GenerateContentRequestBody,
19 mut target: StreamTarget,
20 settings: StreamSettings,
21 state: &GenerationStateAccess<'_, S>,
22 ) -> Result<StreamInvocation<p::ResponsesToChatStream>, TransformError> {
23 let original = input.into_declared();
24
25 let prepared = Self::prepare_with_state(
26 original.clone().buffered(),
27 target.endpoint.clone(),
28 target.identities.clone(),
29 state,
30 )
31 .await?;
32 target.identities = prepared.identities().clone();
33 let bridge = p::ResponsesToChatStream::new_with_policy(
34 target.identities.response.clone(),
35 settings.events.into(),
36 target.identities.response_policy.clone(),
37 )?;
38 let emit_usage = original.emit_usage();
39 let mut invocation = StreamInvocation::new(
40 original,
41 prepared.target_request().clone().streaming(),
42 target,
43 settings,
44 bridge,
45 prepared.report().clone(),
46 state,
47 )
48 .await?;
49 invocation.emit_usage = emit_usage;
50 Ok(invocation)
51 }
52 pub async fn prepare_stream_with_capabilities<S: StateStore, R: ResourceAccess>(
53 input: h::GenerateContentRequestBody,
54 mut target: StreamTarget,
55 settings: StreamSettings,
56 state: &GenerationStateAccess<'_, S>,
57 resources: &GenerationResources<'_, R>,
58 ) -> Result<StreamInvocation<p::ResponsesToChatStream>, TransformError> {
59 let original = input.into_declared();
60
61 let prepared = Self::prepare_with_capabilities(
62 original.clone().buffered(),
63 target.endpoint.clone(),
64 target.identities.clone(),
65 state,
66 resources,
67 )
68 .await?;
69 target.identities = prepared.identities().clone();
70 let bridge = p::ResponsesToChatStream::new_with_policy(
71 target.identities.response.clone(),
72 settings.events.into(),
73 target.identities.response_policy.clone(),
74 )?;
75 let emit_usage = original.emit_usage();
76 let mut invocation = StreamInvocation::new(
77 original,
78 prepared.target_request().clone().streaming(),
79 target,
80 settings,
81 bridge,
82 prepared.report().clone(),
83 state,
84 )
85 .await?;
86 invocation.emit_usage = emit_usage;
87 Ok(invocation)
88 }
89}
90
91impl ResponsesViaChat {
92 pub async fn prepare_stream<S: StateStore>(
93 input: r::GenerateContentRequestBody,
94 target: StreamTarget,
95 context: p::ChatToResponsesContext,
96 settings: StreamSettings,
97 state: &GenerationStateAccess<'_, S>,
98 ) -> Result<StreamInvocation<p::ChatToResponsesStream>, TransformError> {
99 Self::prepare_stream_inner(input, target, context, settings, state, None).await
100 }
101 pub async fn prepare_stream_with_history_cache<S: StateStore>(
102 input: r::GenerateContentRequestBody,
103 target: StreamTarget,
104 context: p::ChatToResponsesContext,
105 settings: StreamSettings,
106 state: &GenerationStateAccess<'_, S>,
107 history_cache: &super::super::ResponsesHistoryCache,
108 ) -> Result<StreamInvocation<p::ChatToResponsesStream>, TransformError> {
109 Self::prepare_stream_inner(input, target, context, settings, state, Some(history_cache))
110 .await
111 }
112 async fn prepare_stream_inner<S: StateStore>(
113 input: r::GenerateContentRequestBody,
114 mut target: StreamTarget,
115 mut context: p::ChatToResponsesContext,
116 settings: StreamSettings,
117 state: &GenerationStateAccess<'_, S>,
118 history_cache: Option<&super::super::ResponsesHistoryCache>,
119 ) -> Result<StreamInvocation<p::ChatToResponsesStream>, TransformError> {
120 let original = input.into_declared();
121 let (history, expanded) = super::super::history::History::prepare_with_cache(
122 &original,
123 state,
124 settings.codec,
125 history_cache,
126 )
127 .await?;
128 context.response.request = original.clone();
129 context.response.request.input = expanded.input.clone();
130 let prepared = Self::prepare_with_state(
131 expanded.buffered(),
132 target.endpoint.clone(),
133 target.identities.clone(),
134 state,
135 )
136 .await?;
137 target.identities = prepared.identities().clone();
138 history.claim_response_id(&mut target.identities, crate::Dialect::OpenAiChat)?;
139
140 let bridge = p::ChatToResponsesStream::new_with_policy(
141 context,
142 target.identities.response.clone(),
143 settings.events.into(),
144 target.identities.response_policy.clone(),
145 )?;
146 let mut invocation = StreamInvocation::new(
147 original,
148 prepared.target_request().clone().streaming(),
149 target,
150 settings,
151 bridge,
152 prepared.report().clone(),
153 state,
154 )
155 .await?;
156 history.bind(&invocation.binding, state)?;
157 invocation.history = Some(history);
158 Ok(invocation)
159 }
160 pub async fn prepare_stream_with_capabilities<S: StateStore, R: ResourceAccess>(
161 input: r::GenerateContentRequestBody,
162 target: StreamTarget,
163 context: p::ChatToResponsesContext,
164 settings: StreamSettings,
165 state: &GenerationStateAccess<'_, S>,
166 resources: &GenerationResources<'_, R>,
167 ) -> Result<StreamInvocation<p::ChatToResponsesStream>, TransformError> {
168 Self::prepare_stream_with_capabilities_inner(
169 input, target, context, settings, state, resources, None,
170 )
171 .await
172 }
173 pub async fn prepare_stream_with_capabilities_and_history_cache<
174 S: StateStore,
175 R: ResourceAccess,
176 >(
177 input: r::GenerateContentRequestBody,
178 target: StreamTarget,
179 context: p::ChatToResponsesContext,
180 settings: StreamSettings,
181 state: &GenerationStateAccess<'_, S>,
182 resources: &GenerationResources<'_, R>,
183 history_cache: &super::super::ResponsesHistoryCache,
184 ) -> Result<StreamInvocation<p::ChatToResponsesStream>, TransformError> {
185 Self::prepare_stream_with_capabilities_inner(
186 input,
187 target,
188 context,
189 settings,
190 state,
191 resources,
192 Some(history_cache),
193 )
194 .await
195 }
196 async fn prepare_stream_with_capabilities_inner<S: StateStore, R: ResourceAccess>(
197 input: r::GenerateContentRequestBody,
198 mut target: StreamTarget,
199 mut context: p::ChatToResponsesContext,
200 settings: StreamSettings,
201 state: &GenerationStateAccess<'_, S>,
202 resources: &GenerationResources<'_, R>,
203 history_cache: Option<&super::super::ResponsesHistoryCache>,
204 ) -> Result<StreamInvocation<p::ChatToResponsesStream>, TransformError> {
205 let original = input.into_declared();
206 let (history, expanded) = super::super::history::History::prepare_with_cache(
207 &original,
208 state,
209 settings.codec,
210 history_cache,
211 )
212 .await?;
213 context.response.request = original.clone();
214 context.response.request.input = expanded.input.clone();
215 let prepared = Self::prepare_with_capabilities(
216 expanded.buffered(),
217 target.endpoint.clone(),
218 target.identities.clone(),
219 state,
220 resources,
221 )
222 .await?;
223 target.identities = prepared.identities().clone();
224 history.claim_response_id(&mut target.identities, crate::Dialect::OpenAiChat)?;
225
226 let bridge = p::ChatToResponsesStream::new_with_policy(
227 context,
228 target.identities.response.clone(),
229 settings.events.into(),
230 target.identities.response_policy.clone(),
231 )?;
232 let mut invocation = StreamInvocation::new(
233 original,
234 prepared.target_request().clone().streaming(),
235 target,
236 settings,
237 bridge,
238 prepared.report().clone(),
239 state,
240 )
241 .await?;
242 history.bind(&invocation.binding, state)?;
243 invocation.history = Some(history);
244 Ok(invocation)
245 }
246}