SolarNetwork/Swarm
2
3
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

✨ Details thinking chunks

Browse Source
This commit is contained in:
LittleSheep
2025-10-26 17:51:08 +08:00
parent c3b6358f33
commit 437f49fb20
11 changed files with 326 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
DysonNetwork.Insight/DysonNetwork.Insight.csproj
View File

@@ -17,6 +17,7 @@
<PackageReference Include="Microsoft.SemanticKernel.Connectors.Ollama" Version="1.66.0-alpha" />
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="9.0.4" />
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL.NodaTime" Version="9.0.4" />
<PackageReference Include="Microsoft.SemanticKernel.Plugins.Web" Version="1.66.0-alpha" />
</ItemGroup>
<ItemGroup>

129
DysonNetwork.Insight/Migrations/20251026045505_AddThinkingChunk.Designer.cs generated Normal file
View File

@@ -0,0 +1,129 @@
// <auto-generated />
using System;
using System.Collections.Generic;
using DysonNetwork.Insight;
using DysonNetwork.Shared.Models;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Migrations;
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
using NodaTime;
using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
#nullable disable
namespace DysonNetwork.Insight.Migrations
{
[DbContext(typeof(AppDatabase))]
[Migration("20251026045505_AddThinkingChunk")]
partial class AddThinkingChunk
{
/// <inheritdoc />
protected override void BuildTargetModel(ModelBuilder modelBuilder)
{
#pragma warning disable 612, 618
modelBuilder
.HasAnnotation("ProductVersion", "9.0.10")
.HasAnnotation("Relational:MaxIdentifierLength", 63);
NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
modelBuilder.Entity("DysonNetwork.Shared.Models.SnThinkingSequence", b =>
{
b.Property<Guid>("Id")
.ValueGeneratedOnAdd()
.HasColumnType("uuid")
.HasColumnName("id");
b.Property<Guid>("AccountId")
.HasColumnType("uuid")
.HasColumnName("account_id");
b.Property<Instant>("CreatedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("created_at");
b.Property<Instant?>("DeletedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("deleted_at");
b.Property<string>("Topic")
.HasMaxLength(4096)
.HasColumnType("character varying(4096)")
.HasColumnName("topic");
b.Property<Instant>("UpdatedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("updated_at");
b.HasKey("Id")
.HasName("pk_thinking_sequences");
b.ToTable("thinking_sequences", (string)null);
});
modelBuilder.Entity("DysonNetwork.Shared.Models.SnThinkingThought", b =>
{
b.Property<Guid>("Id")
.ValueGeneratedOnAdd()
.HasColumnType("uuid")
.HasColumnName("id");
b.Property<List<SnThinkingChunk>>("Chunks")
.IsRequired()
.HasColumnType("jsonb")
.HasColumnName("chunks");
b.Property<string>("Content")
.HasColumnType("text")
.HasColumnName("content");
b.Property<Instant>("CreatedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("created_at");
b.Property<Instant?>("DeletedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("deleted_at");
b.Property<List<SnCloudFileReferenceObject>>("Files")
.IsRequired()
.HasColumnType("jsonb")
.HasColumnName("files");
b.Property<int>("Role")
.HasColumnType("integer")
.HasColumnName("role");
b.Property<Guid>("SequenceId")
.HasColumnType("uuid")
.HasColumnName("sequence_id");
b.Property<Instant>("UpdatedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("updated_at");
b.HasKey("Id")
.HasName("pk_thinking_thoughts");
b.HasIndex("SequenceId")
.HasDatabaseName("ix_thinking_thoughts_sequence_id");
b.ToTable("thinking_thoughts", (string)null);
});
modelBuilder.Entity("DysonNetwork.Shared.Models.SnThinkingThought", b =>
{
b.HasOne("DysonNetwork.Shared.Models.SnThinkingSequence", "Sequence")
.WithMany()
.HasForeignKey("SequenceId")
.OnDelete(DeleteBehavior.Cascade)
.IsRequired()
.HasConstraintName("fk_thinking_thoughts_thinking_sequences_sequence_id");
b.Navigation("Sequence");
});
#pragma warning restore 612, 618
}
}
}

32
DysonNetwork.Insight/Migrations/20251026045505_AddThinkingChunk.cs Normal file
View File

@@ -0,0 +1,32 @@
using System.Collections.Generic;
using DysonNetwork.Shared.Models;
using Microsoft.EntityFrameworkCore.Migrations;
#nullable disable
namespace DysonNetwork.Insight.Migrations
{
/// <inheritdoc />
public partial class AddThinkingChunk : Migration
{
/// <inheritdoc />
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.AddColumn<List<SnThinkingChunk>>(
name: "chunks",
table: "thinking_thoughts",
type: "jsonb",
nullable: false,
defaultValue: new List<SnThinkingChunk>()
);
}
/// <inheritdoc />
protected override void Down(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropColumn(
name: "chunks",
table: "thinking_thoughts");
}
}
}

5
DysonNetwork.Insight/Migrations/AppDatabaseModelSnapshot.cs
View File

@@ -66,6 +66,11 @@ namespace DysonNetwork.Insight.Migrations
.HasColumnType("uuid")
.HasColumnName("id");
b.Property<List<SnThinkingChunk>>("Chunks")
.IsRequired()
.HasColumnType("jsonb")
.HasColumnName("chunks");
b.Property<string>("Content")
.HasColumnType("text")
.HasColumnName("content");

40
DysonNetwork.Insight/Thought/ThoughtController.cs
View File

@@ -1,4 +1,7 @@
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations;
using System.Diagnostics.CodeAnalysis;
using System.IO;
using System.Text;
using System.Text.Json;
using DysonNetwork.Shared.Models;
@@ -21,6 +24,7 @@ public class ThoughtController(ThoughtProvider provider, ThoughtService service)
}
[HttpPost]
[Experimental("SKEXP0110")]
public async Task<ActionResult> Think([FromBody] StreamThinkingRequest request)
{
if (HttpContext.Items["CurrentUser"] is not Account currentUser) return Unauthorized();
@@ -72,7 +76,7 @@ public class ThoughtController(ThoughtProvider provider, ThoughtService service)
// Add previous thoughts (excluding the current user thought, which is the first one since descending)
var previousThoughts = await service.GetPreviousThoughtsAsync(sequence);
var count = previousThoughts.Count;
for (var i = 1; i < count; i++) // skip first (newest, current user)
for (var i = 1; i < count; i++) // skip first (the newest, current user)
{
var thought = previousThoughts[i];
switch (thought.Role)
@@ -99,6 +103,7 @@ public class ThoughtController(ThoughtProvider provider, ThoughtService service)
// Kick off streaming generation
var accumulatedContent = new StringBuilder();
var thinkingChunks = new List<SnThinkingChunk>();
await foreach (var chunk in chatCompletionService.GetStreamingChatMessageContentsAsync(
chatHistory,
provider.CreatePromptExecutionSettings(),
@@ -108,10 +113,33 @@ public class ThoughtController(ThoughtProvider provider, ThoughtService service)
// Process each item in the chunk for detailed streaming
foreach (var item in chunk.Items)
{
var streamingChunk = item switch
{
StreamingTextContent textContent => new SnThinkingChunk
{ Type = StreamingContentType.Text, Data = new() { ["text"] = textContent.Text ?? "" } },
StreamingReasoningContent reasoningContent => new SnThinkingChunk
{
Type = StreamingContentType.Reasoning, Data = new() { ["text"] = reasoningContent.Text ?? "" }
},
StreamingFunctionCallUpdateContent functionCall => new SnThinkingChunk
{
Type = StreamingContentType.FunctionCall,
Data = JsonSerializer.Deserialize<Dictionary<string, object>>(
JsonSerializer.Serialize(functionCall)) ?? new()
},
_ => new SnThinkingChunk
{
Type = StreamingContentType.Unknown, Data = new() { ["data"] = JsonSerializer.Serialize(item) }
}
};
thinkingChunks.Add(streamingChunk);
var messageJson = item switch
{
StreamingTextContent textContent =>
JsonSerializer.Serialize(new { type = "text", data = textContent.Text ?? "" }),
StreamingReasoningContent reasoningContent =>
JsonSerializer.Serialize(new { type = "reasoning", data = reasoningContent.Text ?? "" }),
StreamingFunctionCallUpdateContent functionCall =>
JsonSerializer.Serialize(new { type = "function_call", data = functionCall }),
_ =>
@@ -120,7 +148,7 @@ public class ThoughtController(ThoughtProvider provider, ThoughtService service)
// Write a structured JSON message to the HTTP response as SSE
var messageBytes = Encoding.UTF8.GetBytes($"data: {messageJson}\n\n");
await Response.Body.WriteAsync(messageBytes, 0, messageBytes.Length);
await Response.Body.WriteAsync(messageBytes);
await Response.Body.FlushAsync();
}
@@ -129,8 +157,12 @@ public class ThoughtController(ThoughtProvider provider, ThoughtService service)
}
// Save assistant thought
var savedThought =
await service.SaveThoughtAsync(sequence, accumulatedContent.ToString(), ThinkingThoughtRole.Assistant);
var savedThought = await service.SaveThoughtAsync(
sequence,
accumulatedContent.ToString(),
ThinkingThoughtRole.Assistant,
thinkingChunks
);
// Write the topic if it was newly set, then the thought object as JSON to the stream
using (var streamBuilder = new MemoryStream())

138
DysonNetwork.Insight/Thought/ThoughtProvider.cs
View File

@@ -1,4 +1,5 @@
using System.ClientModel;
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using DysonNetwork.Shared.Models;
using DysonNetwork.Shared.Proto;
@@ -6,8 +7,12 @@ using Microsoft.SemanticKernel;
using Microsoft.SemanticKernel.Connectors.Ollama;
using Microsoft.SemanticKernel.Connectors.OpenAI;
using OpenAI;
using PostPinMode = DysonNetwork.Shared.Proto.PostPinMode;
using PostType = DysonNetwork.Shared.Proto.PostType;
using Microsoft.SemanticKernel.Plugins.Web;
using Microsoft.SemanticKernel.Plugins.Web.Bing;
using Microsoft.SemanticKernel.Plugins.Web.Google;
using NodaTime.Serialization.Protobuf;
using NodaTime.Text;
namespace DysonNetwork.Insight.Thought;
@@ -15,20 +20,26 @@ public class ThoughtProvider
{
private readonly PostService.PostServiceClient _postClient;
private readonly AccountService.AccountServiceClient _accountClient;
private readonly IConfiguration _configuration;
private readonly ILogger<ThoughtProvider> _logger;
public Kernel Kernel { get; }
private string? ModelProviderType { get; set; }
private string? ModelDefault { get; set; }
[Experimental("SKEXP0050")]
public ThoughtProvider(
IConfiguration configuration,
PostService.PostServiceClient postClient,
AccountService.AccountServiceClient accountClient
PostService.PostServiceClient postServiceClient,
AccountService.AccountServiceClient accountServiceClient,
ILogger<ThoughtProvider> logger
)
{
_postClient = postClient;
_accountClient = accountClient;
_logger = logger;
_postClient = postServiceClient;
_accountClient = accountServiceClient;
_configuration = configuration;
Kernel = InitializeThinkingProvider(configuration);
InitializeHelperFunctions();
@@ -63,10 +74,11 @@ public class ThoughtProvider
return builder.Build();
}
[Experimental("SKEXP0050")]
private void InitializeHelperFunctions()
{
// Add Solar Network tools plugin
Kernel.ImportPluginFromFunctions("helper_functions", [
Kernel.ImportPluginFromFunctions("solar_network", [
KernelFunctionFactory.CreateFromMethod(async (string userId) =>
{
var request = new GetAccountRequest { Id = userId };
@@ -80,83 +92,79 @@ public class ThoughtProvider
return JsonSerializer.Serialize(response, GrpcTypeHelper.SerializerOptions);
}, "get_post", "Get a single post by ID from the Solar Network."),
KernelFunctionFactory.CreateFromMethod(async (string query) =>
{
var request = new SearchPostsRequest { Query = query, PageSize = 10 };
var response = await _postClient.SearchPostsAsync(request);
return JsonSerializer.Serialize(response.Posts, GrpcTypeHelper.SerializerOptions);
}, "search_posts", "Search posts by query from the Solar Network."),
{
var request = new SearchPostsRequest { Query = query, PageSize = 10 };
var response = await _postClient.SearchPostsAsync(request);
return JsonSerializer.Serialize(response.Posts, GrpcTypeHelper.SerializerOptions);
}, "search_posts",
"Search posts by query from the Solar Network. The input query is will be used to search with title, description and body content"),
KernelFunctionFactory.CreateFromMethod(async (
string? publisherId = null,
string? realmId = null,
int pageSize = 10,
string? pageToken = null,
string? orderBy = null,
List<string>? categories = null,
List<string>? tags = null,
string? query = null,
List<int>? types = null,
string? afterIso = null,
string? beforeIso = null,
bool includeReplies = false,
string? pinned = null,
bool onlyMedia = false,
bool shuffle = false
string? beforeIso = null
) =>
{
_logger.LogInformation("Begin building request to list post from sphere...");
var request = new ListPostsRequest
{
PublisherId = publisherId,
RealmId = realmId,
PageSize = pageSize,
PageToken = pageToken,
PageSize = 20,
OrderBy = orderBy,
Query = query,
IncludeReplies = includeReplies,
Pinned =
!string.IsNullOrEmpty(pinned) && int.TryParse(pinned, out int p) ? (PostPinMode)p : default,
OnlyMedia = onlyMedia,
Shuffle = shuffle
};
if (!string.IsNullOrEmpty(afterIso))
{
request.After =
Google.Protobuf.WellKnownTypes.Timestamp.FromDateTimeOffset(DateTimeOffset.Parse(afterIso)
.ToUniversalTime());
}
try
{
request.After = InstantPattern.General.Parse(afterIso).Value.ToTimestamp();
}
catch (Exception)
{
_logger.LogWarning("Invalid afterIso format: {AfterIso}", afterIso);
}
if (!string.IsNullOrEmpty(beforeIso))
{
request.Before =
Google.Protobuf.WellKnownTypes.Timestamp.FromDateTimeOffset(DateTimeOffset.Parse(beforeIso)
.ToUniversalTime());
}
try
{
request.Before = InstantPattern.General.Parse(beforeIso).Value.ToTimestamp();
}
catch (Exception)
{
_logger.LogWarning("Invalid beforeIso format: {BeforeIso}", beforeIso);
}
_logger.LogInformation("Request built, {Request}", request);
if (categories != null) request.Categories.AddRange(categories);
if (tags != null) request.Tags.AddRange(tags);
if (types != null) request.Types_.AddRange(types.Select(t => (PostType)t));
var response = await _postClient.ListPostsAsync(request);
return JsonSerializer.Serialize(response.Posts.Select(SnPost.FromProtoValue),
GrpcTypeHelper.SerializerOptions);
var data = response.Posts.Select(SnPost.FromProtoValue);
_logger.LogInformation("Sphere service returned posts: {Posts}", data);
return JsonSerializer.Serialize(data, GrpcTypeHelper.SerializerOptions);
}, "list_posts",
"Get posts from the Solar Network with customizable filters.\n" +
"Get posts from the Solar Network.\n" +
"Parameters:\n" +
"publisherId (optional, string: publisher ID to filter by)\n" +
"realmId (optional, string: realm ID to filter by)\n" +
"pageSize (optional, integer: posts per page, default 20)\n" +
"pageToken (optional, string: pagination token)\n" +
"orderBy (optional, string: field to order by)\n" +
"categories (optional, array of strings: category slugs)\n" +
"tags (optional, array of strings: tag slugs)\n" +
"query (optional, string: search query, will search in title, description and body)\n" +
"types (optional, array of integers: post types, use 0 for Moment, 1 for Article)\n" +
"orderBy (optional, string: order by published date, accept asc or desc)\n" +
"afterIso (optional, string: ISO date for posts after this date)\n" +
"beforeIso (optional, string: ISO date for posts before this date)\n" +
"includeReplies (optional, boolean: include replies, default false)\n" +
"pinned (optional, string: pin mode as integer string, '0' for PublisherPage, '1' for RealmPage, '2' for ReplyPage)\n" +
"onlyMedia (optional, boolean: only posts with media, default false)\n" +
"shuffle (optional, boolean: shuffle results, default false)"
"beforeIso (optional, string: ISO date for posts before this date)"
)
]);
// Add web search plugins if configured
var bingApiKey = _configuration.GetValue<string>("Thinking:BingApiKey");
if (!string.IsNullOrEmpty(bingApiKey))
{
var bingConnector = new BingConnector(bingApiKey);
var bing = new WebSearchEnginePlugin(bingConnector);
Kernel.ImportPluginFromObject(bing, "bing");
}
var googleApiKey = _configuration.GetValue<string>("Thinking:GoogleApiKey");
var googleCx = _configuration.GetValue<string>("Thinking:GoogleCx");
if (!string.IsNullOrEmpty(googleApiKey) && !string.IsNullOrEmpty(googleCx))
{
var googleConnector = new GoogleConnector(
apiKey: googleApiKey,
searchEngineId: googleCx);
var google = new WebSearchEnginePlugin(googleConnector);
Kernel.ImportPluginFromObject(google, "google");
}
}
public PromptExecutionSettings CreatePromptExecutionSettings()

16
DysonNetwork.Insight/Thought/ThoughtService.cs
View File

@@ -6,7 +6,8 @@ namespace DysonNetwork.Insight.Thought;
public class ThoughtService(AppDatabase db, ICacheService cache)
{
public async Task<SnThinkingSequence?> GetOrCreateSequenceAsync(Guid accountId, Guid? sequenceId, string? topic = null)
public async Task<SnThinkingSequence?> GetOrCreateSequenceAsync(Guid accountId, Guid? sequenceId,
string? topic = null)
{
if (sequenceId.HasValue)
{
@@ -23,13 +24,19 @@ public class ThoughtService(AppDatabase db, ICacheService cache)
}
}
public async Task<SnThinkingThought> SaveThoughtAsync(SnThinkingSequence sequence, string content, ThinkingThoughtRole role)
public async Task<SnThinkingThought> SaveThoughtAsync(
SnThinkingSequence sequence,
string content,
ThinkingThoughtRole role,
List<SnThinkingChunk>? chunks = null
)
{
var thought = new SnThinkingThought
{
SequenceId = sequence.Id,
Content = content,
Role = role
Role = role,
Chunks = chunks ?? []
};
db.ThinkingThoughts.Add(thought);
await db.SaveChangesAsync();
@@ -60,7 +67,8 @@ public class ThoughtService(AppDatabase db, ICacheService cache)
return thoughts;
}
public async Task<(int total, List<SnThinkingSequence> sequences)> ListSequencesAsync(Guid accountId, int offset, int take)
public async Task<(int total, List<SnThinkingSequence> sequences)> ListSequencesAsync(Guid accountId, int offset,
int take)
{
var query = db.ThinkingSequences.Where(s => s.AccountId == accountId);
var totalCount = await query.CountAsync();

6
DysonNetwork.Insight/appsettings.json
View File

@@ -20,8 +20,8 @@
"Insecure": true
},
"Thinking": {
"Provider": "ollama",
"Model": "qwen3:8b",
"Endpoint": "http://localhost:11434/api"
"Provider": "deepseek",
"Model": "deepseek-chat",
"ApiKey": "sk-cd709f9f1b96432e99d2d992392b4220"
}
}

16
DysonNetwork.Shared/Models/ThinkingSequence.cs
View File

@@ -18,6 +18,20 @@ public enum ThinkingThoughtRole
User
}
public enum StreamingContentType
{
Text,
Reasoning,
FunctionCall,
Unknown
}
public class SnThinkingChunk
{
public StreamingContentType Type { get; set; }
public Dictionary<string, object>? Data { get; set; } = new();
}
public class SnThinkingThought : ModelBase
{
public Guid Id { get; set; } = Guid.NewGuid();
@@ -25,6 +39,8 @@ public class SnThinkingThought : ModelBase
[Column(TypeName = "jsonb")] public List<SnCloudFileReferenceObject> Files { get; set; } = [];
[Column(TypeName = "jsonb")] public List<SnThinkingChunk> Chunks { get; set; } = [];
public ThinkingThoughtRole Role { get; set; }
public Guid SequenceId { get; set; }

11
DysonNetwork.Shared/Proto/GrpcTypeHelper.cs
View File

@@ -51,6 +51,7 @@ public abstract class GrpcTypeHelper
_ => Value.ForString(JsonSerializer.Serialize(kvp.Value, SerializerOptions)) // fallback to JSON string
};
}
return result;
}
@@ -66,13 +67,15 @@ public abstract class GrpcTypeHelper
try
{
// Try to parse as JSON object or primitive
result[kvp.Key] = JsonNode.Parse(value.StringValue)?.AsObject() ?? JsonObject.Create(new JsonElement());
result[kvp.Key] = JsonNode.Parse(value.StringValue)?.AsObject() ??
JsonObject.Create(new JsonElement());
}
catch
{
// Fallback to raw string
result[kvp.Key] = value.StringValue;
}
break;
case Value.KindOneofCase.NumberValue:
result[kvp.Key] = value.NumberValue;
@@ -106,6 +109,7 @@ public abstract class GrpcTypeHelper
break;
}
}
return result;
}
@@ -117,7 +121,8 @@ public abstract class GrpcTypeHelper
Value.KindOneofCase.NumberValue => value.NumberValue,
Value.KindOneofCase.BoolValue => value.BoolValue,
Value.KindOneofCase.NullValue => null,
_ => JsonSerializer.Deserialize<JsonElement>(JsonSerializer.Serialize(value, SerializerOptions), SerializerOptions)
_ => JsonSerializer.Deserialize<JsonElement>(JsonSerializer.Serialize(value, SerializerOptions),
SerializerOptions)
};
}
@@ -165,6 +170,6 @@ public abstract class GrpcTypeHelper
public static T? ConvertByteStringToObject<T>(ByteString bytes)
{
return JsonSerializer.Deserialize<T>(bytes.ToStringUtf8(), SerializerOptions);
return bytes.IsEmpty ? default : JsonSerializer.Deserialize<T>(bytes.ToStringUtf8(), SerializerOptions);
}
}

6
DysonNetwork.Sphere/Post/PostServiceGrpc.cs
View File

@@ -127,7 +127,11 @@ public class PostServiceGrpc(AppDatabase db, PostService ps) : Shared.Proto.Post
query = request.Shuffle
? query.OrderBy(e => EF.Functions.Random())
: query.OrderByDescending(e => e.PublishedAt ?? e.CreatedAt);
: request.OrderBy switch
{
"asc" => query.OrderBy(e => e.PublishedAt ?? e.CreatedAt),
_ => query.OrderByDescending(e => e.PublishedAt ?? e.CreatedAt),
};
if (!string.IsNullOrWhiteSpace(request.PublisherId) && Guid.TryParse(request.PublisherId, out var pid))
query = query.Where(p => p.PublisherId == pid);