MCPcopy Create free account
hub / github.com/chainreactors/EvilProxy / executeStreamWithModelPool

Method executeStreamWithModelPool

sdk/cliproxy/auth/conductor.go:1821–1915  ·  view source on GitHub ↗
(ctx context.Context, executor ProviderExecutor, auth *Auth, provider string, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, routeModel string, execModels []string, pooled bool, aliasResult OAuthModelAliasResult)

Source from the content-addressed store, hash-verified

1819}
1820
1821func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor ProviderExecutor, auth *Auth, provider string, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, routeModel string, execModels []string, pooled bool, aliasResult OAuthModelAliasResult) (*cliproxyexecutor.StreamResult, error) {
1822 if executor == nil {
1823 return nil, &Error{Code: "executor_not_found", Message: "executor not registered"}
1824 }
1825 ctx = contextWithRequestedModelAlias(ctx, opts, routeModel)
1826 var lastErr error
1827 for idx, execModel := range execModels {
1828 resultModel := m.stateModelForExecution(auth, routeModel, execModel, pooled)
1829 execReq := req
1830 execReq.Model = execModel
1831 execOpts := opts
1832 execReq, execOpts = applyRequestAfterAuthInterceptor(ctx, executor, provider, execReq, execOpts, requestedModelAliasFromOptions(execOpts, routeModel))
1833 streamResult, errStream := executor.ExecuteStream(ctx, auth, execReq, execOpts)
1834 if errStream != nil {
1835 if errCtx := ctx.Err(); errCtx != nil {
1836 return nil, errCtx
1837 }
1838 rerr := &Error{Message: errStream.Error()}
1839 if se, ok := errors.AsType[cliproxyexecutor.StatusError](errStream); ok && se != nil {
1840 rerr.HTTPStatus = se.StatusCode()
1841 }
1842 result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr}
1843 result.RetryAfter = retryAfterFromError(errStream)
1844 m.MarkResult(ctx, result)
1845 if isRequestInvalidError(errStream) {
1846 return nil, errStream
1847 }
1848 lastErr = errStream
1849 continue
1850 }
1851
1852 buffered, closed, bootstrapErr := readStreamBootstrap(ctx, streamResult.Chunks)
1853 if bootstrapErr != nil {
1854 if errCtx := ctx.Err(); errCtx != nil {
1855 discardStreamChunks(streamResult.Chunks)
1856 return nil, errCtx
1857 }
1858 if isRequestInvalidError(bootstrapErr) {
1859 rerr := &Error{Message: bootstrapErr.Error()}
1860 if se, ok := errors.AsType[cliproxyexecutor.StatusError](bootstrapErr); ok && se != nil {
1861 rerr.HTTPStatus = se.StatusCode()
1862 }
1863 result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr}
1864 result.RetryAfter = retryAfterFromError(bootstrapErr)
1865 m.MarkResult(ctx, result)
1866 discardStreamChunks(streamResult.Chunks)
1867 return nil, bootstrapErr
1868 }
1869 if idx < len(execModels)-1 {
1870 rerr := &Error{Message: bootstrapErr.Error()}
1871 if se, ok := errors.AsType[cliproxyexecutor.StatusError](bootstrapErr); ok && se != nil {
1872 rerr.HTTPStatus = se.StatusCode()
1873 }
1874 result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr}
1875 result.RetryAfter = retryAfterFromError(bootstrapErr)
1876 m.MarkResult(ctx, result)
1877 discardStreamChunks(streamResult.Chunks)
1878 lastErr = bootstrapErr

Calls 15

MarkResultMethod · 0.95
wrapStreamResultMethod · 0.95
retryAfterFromErrorFunction · 0.85
isRequestInvalidErrorFunction · 0.85
readStreamBootstrapFunction · 0.85
discardStreamChunksFunction · 0.85
newStreamBootstrapErrorFunction · 0.85
ExecuteStreamMethod · 0.65

Tested by

no test coverage detected