schema_provider.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/provider.h>
#include <neograph/llm/json_path.h>
#include <asio/executor_work_guard.hpp>
#include <asio/io_context.hpp>
#include <fstream>
#include <memory>
#include <mutex>
#include <optional>
#include <set>
#include <string>
#include <thread>
#include <map>
 
namespace neograph::async { class ConnPool; class CurlH2Pool; }
 
namespace neograph::llm {
 
namespace test_access { class SchemaProviderTestAccess; }  // fwd-decl for friend
 
class NEOGRAPH_API SchemaProvider : public Provider {
  public:
    struct Config {
        std::string schema_path;                    
        std::string api_key;                        
        std::string default_model = "gpt-4o-mini";  
        int timeout_seconds = 60;                   
        std::string base_url_override;              
 
        bool use_websocket = false;
 
        bool prefer_libcurl = false;
    };
 
    static std::unique_ptr<SchemaProvider> create(const Config& config);
 
    ~SchemaProvider();
 
    asio::awaitable<ChatCompletion>
    complete_async(const CompletionParams& params) override;
 
 
    ChatCompletion complete_stream(const CompletionParams& params,
                                   const StreamCallback& on_chunk) override;
 
    asio::awaitable<ChatCompletion>
    complete_stream_async(const CompletionParams& params,
                          const StreamCallback& on_chunk) override;
 
    asio::awaitable<ChatCompletion>
    invoke(const CompletionParams& params, StreamCallback on_chunk) override;
 
    std::string get_name() const override;
 
  private:
    explicit SchemaProvider(Config config, json schema);
 
    // --- Strategies (internal) ---
    enum class SystemPromptStrategy { IN_MESSAGES, TOP_LEVEL, TOP_LEVEL_PARTS };
    // FLAT_ITEMS: OpenAI Responses — tool calls are separate top-level items in input[] (not nested in a message).
    enum class ToolCallStrategy { TOOL_CALLS_ARRAY, CONTENT_ARRAY, PARTS_ARRAY, FLAT_ITEMS };
    // FLAT_ITEM: OpenAI Responses — {type:"function_call_output", call_id, output} as a top-level input[] item.
    enum class ToolResultStrategy { FLAT, CONTENT_ARRAY, PARTS_ARRAY, FLAT_ITEM };
    // FLAT_FUNCTION: OpenAI Responses — [{type:"function", name, description, parameters}] (no nesting under "function").
    enum class ToolDefWrapper { FUNCTION, NONE, FUNCTION_DECLARATIONS, FLAT_FUNCTION };
    // OUTPUT_ARRAY: OpenAI Responses — output[] with mixed message/function_call items.
    enum class ResponseStrategy { CHOICES_MESSAGE, CONTENT_ARRAY, CANDIDATES_PARTS, OUTPUT_ARRAY };
    enum class StreamFormat { SSE_DATA, SSE_EVENTS };
 
    // --- Internal config parsed from schema ---
    struct ConnectionConfig {
        std::string base_url;
        std::string endpoint;
        std::string stream_endpoint;
        std::string auth_header;
        std::string auth_prefix;
        std::string api_key_env;
        std::string auth_query_param;
        std::map<std::string, std::string> extra_headers;
    };
 
    struct RequestConfig {
        std::string model_field;
        std::string messages_field;
        std::string tools_field;
        std::string temperature_path;
        std::string max_tokens_path;
        bool max_tokens_required = false;
        int max_tokens_default = -1;
        std::string stream_field;
        json extra_fields;
 
        std::set<std::string> per_call_fields;
    };
 
    struct SystemPromptConfig {
        SystemPromptStrategy strategy;
        std::string field;
        std::string role_name;
        std::string parts_field;
        std::string text_field;
    };
 
    struct MessagesConfig {
        std::string role_field;
        std::string content_field;
        std::map<std::string, std::string> role_map;
        bool content_is_parts = false;
        json text_part_template;
    };
 
    struct ToolDefConfig {
        ToolDefWrapper wrapper;
        std::string name_field;
        std::string description_field;
        std::string parameters_field;
    };
 
    struct ToolCallConfig {
        ToolCallStrategy strategy;
        std::string field;
        json item_template;
        json text_item_template;
    };
 
    struct ToolResultConfig {
        std::string role;
        ToolResultStrategy strategy;
        std::string id_field;
        std::string content_field;
        json item_template;
    };
 
    struct ImageConfig {
        std::string strategy;
        json item_template;
        json text_part_template;
    };
 
    struct ResponseConfig {
        ResponseStrategy strategy;
        std::string message_path;
        std::string content_field;
        std::string role_field;
        std::string tool_calls_field;
        std::string tool_call_id_field;
        std::string tool_call_name_path;
        std::string tool_call_args_path;
        bool tool_call_args_is_string = true;
        std::string content_path;
        std::string text_type;
        std::string text_field;
        std::string tool_use_type;
        std::string tool_call_name_field;
        std::string tool_call_args_field;
        std::string parts_path;
        std::string function_call_field;
        // OUTPUT_ARRAY (OpenAI Responses)
        std::string output_path;             
        std::string message_item_type;       
        std::string function_call_item_type; 
        std::string message_content_field;   
        std::string function_call_id_field;  
        std::string usage_path;
        std::string prompt_tokens_field;
        std::string completion_tokens_field;
        std::string total_tokens_field;
    };
 
    struct StreamConfig {
        StreamFormat format;
        std::string prefix;
        std::string done_signal;
        std::string delta_path;
        std::string content_field;
        std::string tool_calls_field;
        std::string tool_call_index_field;
        std::string tool_call_id_field;
        std::string tool_call_name_path;
        std::string tool_call_args_path;
        std::string delta_strategy;
        std::string delta_parts_path;
        std::string delta_text_field;
        std::string delta_function_call_field;
        std::string delta_tool_call_name_field;
        std::string delta_tool_call_args_field;
        json events_config;
    };
 
    // Serializes access to the schema-derived json templates (schema_,
    // tool_call_.item_template, tool_result_.item_template, req_.extra_fields,
    // etc.). These are backed by shared yyjson_mut_doc handles — even
    // read-only traversal of a yyjson_mut_val from multiple threads at once
    // trips internal iterator state that yyjson explicitly disclaims as
    // thread-unsafe for mutable docs. HTTP I/O is issued OUTSIDE this lock
    // so concurrent fan-out requests still overlap on the network.
    mutable std::mutex schema_mutex_;
 
    // --- Connection pool for HTTP keep-alive ---
    //
    // Each Provider::complete() goes through run_sync, which creates a
    // fresh asio::io_context per call. A ConnPool bound to that
    // throw-away executor would survive only one request — defeating
    // its purpose. So SchemaProvider owns its own long-lived
    // io_context + worker thread; the pool is bound to that, and
    // every complete_async dispatches through it. Successive calls to
    // the same host then amortise TCP connect + TLS handshake.
    std::unique_ptr<asio::io_context> http_io_;
    std::optional<asio::executor_work_guard<asio::io_context::executor_type>> http_work_;
    std::thread http_thread_;
    std::unique_ptr<async::ConnPool>    conn_pool_;
 
    // --- Long-lived "sync-bridge" thread for streaming (issue #16) ---
    //
    // The streaming HTTP/SSE path is implemented as a synchronous
    // httplib::Client::Post call inside `complete_stream`. The previous
    // `complete_stream_async` default ran that on a *fresh* `std::thread`
    // per call, which exposed cold thread-local resolver / NSS init in
    // glibc. The wild ptr in `internal_strlen` reported in #16 had this
    // shape: outer io.run() driven from an HTTP server worker thread →
    // fresh-spawn NeoGraph worker → first getaddrinfo on cold TLS.
    //
    // Fix: own one long-lived bridge thread (mirror of `http_thread_`
    // for ConnPool). `complete_stream_async` HTTP/SSE branch dispatches
    // each call onto this thread instead of spawning fresh. After the
    // first call warms the thread-local resolver state, every
    // subsequent call reuses the warm state — same robustness profile
    // as the working `complete_async` path.
    std::unique_ptr<asio::io_context> bridge_io_;
    std::optional<asio::executor_work_guard<asio::io_context::executor_type>> bridge_work_;
    std::thread bridge_thread_;
    // libcurl-backed HTTP/2 pool with multiplexing. Default transport
    // for SchemaProvider — passes Cloudflare/anti-bot WAFs (it IS curl)
    // and gives us native HTTP/2 stream multiplexing for parallel
    // fan-out workloads.
    std::unique_ptr<async::CurlH2Pool>  curl_pool_;
 
    // --- Parsed config ---
    Config user_config_;
    json schema_;
    std::string provider_name_;
    ConnectionConfig conn_;
    RequestConfig req_;
    SystemPromptConfig sys_;
    MessagesConfig msgs_;
    ToolDefConfig tool_def_;
    ToolCallConfig tool_call_;
    ToolResultConfig tool_result_;
    ImageConfig image_;
    ResponseConfig resp_;
    StreamConfig stream_;
 
    // --- Internal methods ---
    void parse_schema();
 
    json build_body(const CompletionParams& params) const;
    json serialize_messages(const std::vector<ChatMessage>& messages) const;
    json serialize_tools(const std::vector<ChatTool>& tools) const;
    json serialize_single_message(const ChatMessage& msg) const;
 
    asio::awaitable<ChatCompletion>
    complete_stream_ws_responses(const CompletionParams& params,
                                 const StreamCallback& on_chunk);
 
    ChatMessage parse_response(const json& resp_json) const;
    ChatCompletion::Usage parse_usage(const json& resp_json) const;
 
    std::string build_endpoint(const std::string& model, bool streaming) const;
    std::map<std::string, std::string> build_headers() const;
    std::string get_api_key() const;
 
    static std::pair<std::string, std::string> parse_data_url(const std::string& url);
    static json substitute(const json& tmpl, const std::map<std::string, json>& vars);
    static std::string generate_tool_call_id();
 
    // Test-only access to private internals. The helper class lives in a
    // separate `test_access` namespace to discourage accidental use; tests
    // that need to inspect build_body / serialize_messages output for
    // contract verification (e.g. issue #34, #35 regression coverage)
    // pull it in explicitly. NOT a public API surface — may change without
    // notice between versions.
    friend class neograph::llm::test_access::SchemaProviderTestAccess;
};
 
namespace test_access {
 
class SchemaProviderTestAccess {
  public:
    static json build_body(const SchemaProvider& sp,
                           const CompletionParams& params) {
        return sp.build_body(params);
    }
};
 
}  // namespace test_access
 
} // namespace neograph::llm