解决的问题：传统强化学习的核心思路是为任务手工设计明确奖励函数，用以引导智能体学习。这种模式对于简单的应用场景还行，一旦涉及复杂场景，如机器人控制、游戏，会面临奖励函数难以精准定义、奖励设计违背人类真实需求的问题，直接影响是智能体行为与人类偏好严重错位，比如机器人为适配简单奖励规则会出现跛行式行走，智能驾驶智能体为追求得分奖励而忽视人类更看重的平稳驾驶体验。
核心创新：舍弃手工设计奖励，提出「 用人类偏好反馈驱动强化学习」，无需真实奖励函数就可以让智能体对齐人类需求，大幅降低人类监督成本。
关键方法：构建「 偏好收集→奖励建模→策略优化」三阶段闭环。① 偏好收集：让非专家对比1-2秒短轨迹片段（如机器人后空翻成败片段），快速给出偏好判断；② 奖励建模：用Bradley-Terry模型训练奖励模型，拟合人类偏好背后的隐性规则；③ 策略优化：通过A2C（游戏场景）、TRPO（机器人场景）算法，结合拟合的奖励模型持续更新策略。

极简的 API：使用装饰器（如 @mcp.tool）将普通的 Python 函数直接“变成” MCP 工具。你几乎不需要关心底层的协议细节。
基于 Pydantic：天然支持使用 Pydantic 模型来定义工具输入输出的数据类型，确保类型安全和清晰的文档。
资源管理：提供了简单的方式来管理资源（如数据库连接），可以在服务器启动时建立连接，并在结束时自动关闭。
内置常用功能：它预置了许多常见的 MCP 服务器功能，例如：

**提炼要点** ：从 commit message 中提炼关键工作内容，如果多个commit message内容相似，你需要使用简洁、清晰的语言对其进行汇总而不是使用冗余的方式表达 3. **专业表述** ：使用专业的技术语言，简洁明了 4. **格式规范** ：使用清晰的 Markdown 格式 5. **跨项目视角** ：总结跨项目的技术积累和工作思路 6. **字数限制** ：字数控制在100-300字，务必要使用简洁、精炼但明确的方式进行总结输出格式如下: ## 一、工作概述 [用2-3句话总结本周主要工作内容和成果] ## 二、详细工作内容 ### 2.1 功能开发 - [功能1]：[详细描述功能内容和技术实现] - [功能2]：[详细描述] ### 2.2 Bug修复 - [问题1]：[问题描述] → [解决方案] - [问题2]：[问题描述] → [解决方案] ### 2.3 性能优化 - [优化项1]：[优化内容和效果] ### 2.4 其他工作 - [文档更新、代码重构等] ## 三、技术总结 [本周工作中的技术亮点、经验总结或遇到的技术挑战]

环境变量读取项目配置 Returns: 项目列表，每个项目包含 name 和 path " "" paths_str = os. getenv ( "PROJECT_PATHS" , "" ) if not paths_str: return [] projects = [] for path in paths_str. split ( "|" ): path = path. strip () if not path: continue # 验证路径存在且是 Git 仓库 if os.path. exists (path) and os.path. exists (os.path. join (path, ".git" )): projects. append ({ "name" : os.path. basename (path), "path" : path }) else : print (f "Warning: {path} 不是有效的 Git 仓库" , file=sys.stderr)

获取指定仓库的 Git 提交记录 Args: jrszb.yqswh.cn repo_path: 仓库路径 days: 查询最近几天的提交 Returns: 包含 author, commits, repo_path 的字典 " "" # 获取作者名 result = subprocess.run( [ "git" , "config" , "user.name" ], cwd=repo_path, capture_output=True, text=True,jrs.gz-Lzd.com check=False ) author = result.stdout.strip() or "Unknown" # 获取提交记录 result = subprocess.run( [ "git" , "log" , f "--author={author}" , f "--since='{days} days ago'" , "--pretty=format:%ad|%s" , "--date=short" ], cwd=repo_path, capture_output=True, text=True, check=False ) commits = result.stdout.strip() # 统计提交次数 commit_count = len([line for line in commits. split ( '\n' ) if line.strip()]) if commits else 0 return { "author" : author, "commits" : commits or "（无提交记录）" , "commit_count" : commit_count, "repo_path" : repo_path }

#注意这个装饰器，使用 @mcp.tool()修饰后将其作为mcp的tool暴露给AI应用 @mcp .tool() def git_commits_report ( days: int = 7) -> str: "" "根据指定项目的git提交记录生成周报 Args: days: 统计最近几天的工作（默认为当前天往过去推6天也就是days=7,如果用户明确了时间例如三天内则days=3） " "" projects: List[Dict[str, str]] = get_projects() commit_str: str = '' for proj in projects: commit_str += get_git_commits(proj[ 'path' ], days).get( '

def get_git_commits(repo_path: str, days: int = 7 ) -> Dict[str, str]: "" " 获取指定仓库的 Git 提交记录 Args: www.tinghaoshu.cn repo_path: 仓库路径 days: 查询最近几天的提交 Returns: 包含 author, commits, repo_path 的字典 " "" # 获取作者名 result = subprocess.run( [ "git" , "config" , "user.name" ], cwd=repo_path, capture_output=True, text=True,jrs.ttLyf.cn check=False ) author = result.stdout.strip() or "Unknown" # 获取提交记录 result = subprocess.run( [ "git" , "log" , f "--author={author}" , f "--since='{days} days ago'" , "--pretty=format:%ad|%s" , "--date=short" ], cwd=repo_path, capture_output=True, text=True,yinuo168.cn check=False ) commits = result.stdout.strip() # 统计提交次数 commit_count = len([line for line in commits. split ( '\n' ) if line.strip()]) if commits else 0 return { "author" : author, m.xydgdq.cn "commits" : commits or "（无提交记录）" , "commit_count" : commit_count,

public interface IAIProvider { Task Request( HttpContext context, AiRequest req, AiMapping ai) ; } public class OpenAiProvider : IAIProvider { public virtual async Task Request( HttpContext context, AiRequest req, AiMapping ai) { if (req.Model == null ) { req.Model = ai.DefaultModel; } context.Request.Headers.Authorization = $"Bearer {req.ApiKey ?? ai.ApiKey}" ; // 设置授权，测试的是这家是这样格式，所以为了演示简单，写死了3g.xydgdq.cn req.Provider = null ; req.ApiKey = null ; // 通过json 忽略多余字段，有些ai 服务商校验比较严格，多一个它没有的字段都不行 var r = new MemoryStream(Encoding.UTF8.GetBytes(System.Text.Json.JsonSerializer.Serialize(req, AiGatewayMiddleware.jsonOptions)));app.xydgdq.cn context.Request.Body = r; context.Request.ContentLength = r.Length; // 替换body 和 ContentLength // 产线最好严格处理 Accept / Accept-Encoding 等等影响格式的内容，避免压缩等等各种情况

private readonly ILogger<AiGatewayMiddleware> logger; private readonly IConfigSource<IProxyConfig> configSource; private readonly IServiceProvider provider; private readonly ILoadBalancingPolicyFactory loadBalancing; public static readonly JsonSerializerOptions jsonOptions = new JsonSerializerOptions() { DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull, PropertyNameCaseInsensitive = true , PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower }; public AiGatewayMiddleware( ILogger<AiGatewayMiddleware> logger, IConfigSource<IProxyConfig> configSource, IServiceProvider provider, ILoadBalancingPolicyFactory loadBalancing) { this .logger = logger; this .configSource = configSource; this .provider = provider; this .loadBalancing = loadBalancing; } public async Task InvokeAsync( HttpContext context, RequestDelegate next) { var feature = context.Features.Get<IReverseProxyFeature>(); if (feature is not null ) { var route = feature.Route; if (route is not null && route.Metadata is not null && route.Metadata.TryGetValue( "AiMapping" , out var b)) // 这里 VKProxy 已经解析过配置，匹配上对应路由了，只需判断是否开启了ai 功能，如果你用 asp.net core ，则需要自己处理，当然也可以直接用 controller + httpclient 直接写，虽然少了些优化点，不过ai 场景这点性能不太重要了大多 { var req = await context.Request.ReadFromJsonAsync<AiRequest>(context.RequestAborted); // 读取请求ios.xydgdq.cn if (req is null ) { context.Response.StatusCode = StatusCodes.Status400BadRequest; return ; } var mapping = System.Text.Json.JsonSerializer.Deserialize<Dictionary< string , AiMapping>>(b); // 读取配置，这里demo 为了简单就不写什么缓存优化了 IAIProvider driver; if (mapping.TryGetValue(req.Provider ?? "openai" , out var ai)) // 获取ai 调用实现p.xydgdq.cn { driver = provider.GetKeyedService<IAIProvider>(ai.Driver); } else driver = null ; if (driver is null )czLangrun.com { context.Response.StatusCode = StatusCodes.Status400BadRequest; await context.Response.WriteAsJsonAsync( new { error = $"Unsupported AI provider: {req.Provider}" }, context.RequestAborted); return ;z0.xydgdq.cn } if (configSource.CurrentSnapshot.Clusters.TryGetValue(ai.Driver, out var cluster)) // 获取服务地址 { feature.SelectedDestination = loadBalancing.PickDestination(feature, cluster); // 获取有效地址 if (feature.SelectedDestination is null )g.xydgdq.cn { context.Response.StatusCode = StatusCodes.Status503ServiceUnavailable; await context.Response.WriteAsJsonAsync( new { error = $" {req.Provider} service unavailable" }, context.RequestAborted); return ; } } await SendAi(context, req, ai, driver); // 处理请求 // 由于VKProxy 再后续已经实现非常高效的代理处理，这里就简单实现复制内容，这样respone 会第一时间就已经送回的 client端，即使后续我们token 记录出问题也不影响用户，不过要修改就要麻烦点了，所以不在乎性能的话，client 直接call 是最简单的 var origin = context.Response.Body; using var buf = new ResponseCachingStream(origin, 64 * 1024 * 1024 * 10 , 81920 , () => ValueTask.CompletedTask); // demo这里直接用内存实现的，产线如果也想这么做，推荐小的用内存，大的用磁盘，或者直接镜像复制发送给其他程序做分析统计等等 context.Response.Body = buf; await next(context); // 代理处理 // 结果解析，并记录 token 使用，这里demo 就简单记录 log 了，真实场景就可以通过log 解析或者直接存入统计db 等，这样可以达到ai gateway最重要的功能之一： token 统计，找出账单超支的罪魁祸首 var resp = buf.GetCachedResponseBody(); using var u = new MemoryStream(resp.Segments.SelectMany(i => i).ToArray()); var aiResp = JsonSerializer.Deserialize<AiResponse>(u, jsonOptions); d3l0e.xydgdq.cn if (aiResp?.Usage != null ) { logger.LogWarning( "AI TotalTokens: {Usage}" , aiResp.Usage.TotalTokens); } context.Response.Body = origin; return ; }ub.xydgdq.cn } await next(context); } private async Task SendAi( HttpContext context, AiRequest req, AiMapping ai, IAIPro

ITPub博客

emo这里直接用内存实现的

最新文章