1. 为什么我们需要一个“小而清晰”的Agent Runtime最近在折腾AI Agent项目时我遇到了一个非常典型的问题项目初期为了快速验证想法我直接上手了LangChain。它确实强大开箱即用各种工具链和记忆模块一应俱全。但随着项目深入我需要将Agent部署到一个资源受限的边缘设备上问题就来了。LangChain的依赖树庞大启动慢内存占用高而且它内部对OpenAI API的调用逻辑是“硬编码”的我想切换到本地部署的Qwen或者通义千问模型改动起来异常麻烦几乎要重写整个调用链路。这让我开始思考我们真的需要一个如此“重量级”的框架来运行一个Agent吗一个Agent的核心运行时Runtime到底应该提供什么在我看来它最核心的职责无非是这几件事管理对话状态记忆、按计划调用工具或模型、处理工具返回结果、并根据结果决定下一步行动。至于具体用哪个大模型、工具怎么实现、记忆存到哪里这些应该是可插拔、可替换的“策略”而不是被框架绑死的“实现”。这就是“Runlet”这个想法诞生的背景。我想构建一个Python的Agent运行时它的设计哲学就是“小而清晰Provider Neutral”。“小而清晰”意味着核心逻辑极其精简没有不必要的抽象层代码一目了然易于调试和定制。“Provider Neutral”提供商中立则是更关键的一点它要求运行时核心完全不关心你用的是OpenAI、Anthropic的Claude、还是本地部署的Llama、Qwen。它只定义一套统一的接口你注入什么模型实现它就用什么。这带来了巨大的灵活性无论是云端API还是本地模型无论是商业服务还是开源项目都可以无缝接入。从网络热词也能看出大家的痛点claude显示your connection works, but the provider rejected a test request是典型的提供商接口问题oai compatible provider for copilot和cursor使用本地模型qwen provider returned error则反映了开发者对兼容不同模型提供商的强烈需求而wmi provider host占用高、cannot find any provider supporting aes/cbc/pkcs7padding这类错误虽然领域不同但都揭示了“运行时”与“提供者”之间松耦合的重要性——一个部分出错不应导致整个系统崩溃。所以Runlet的目标不是成为另一个功能大而全的Agent框架而是成为一个坚实、透明、无绑定的底层“发动机”。你可以基于它快速构建原型也可以将它嵌入到对性能和依赖有严格要求的生产环境中而不必担心被某个特定的云服务或模型框架“锁死”。2. 拆解Runlet的核心架构一个极简的状态机要实现“小而清晰”首要任务是界定边界做最核心的事。Runlet将Agent的一次执行抽象为一个状态机。这个状态机非常简洁只围绕几个核心状态和动作运转。2.1 核心状态定义一个Agent在Runlet中的生命周期主要由以下几个状态构成IDLE空闲Agent的初始状态等待执行指令。THINKING思考Agent正在调用语言模型LLM进行“思考”生成下一步的行动计划可能是一个工具调用也可能是最终答复。ACTING执行Agent正在调用一个具体的工具如搜索网络、执行计算、查询数据库并等待结果。OBSERVING观察工具执行完毕Agent收到了结果正在处理这个结果并准备决定下一步。FINISHED完成Agent任务完成输出了最终结果。ERROR错误执行过程中发生不可恢复的错误。这个状态机是同步的、单线程的它不处理并发因为那是上层调度器该做的事。Runlet只保证一次执行的正确流转。2.2 核心组件与接口为了驱动这个状态机Runlet定义了三个最核心的抽象接口这也是“Provider Neutral”设计的关键所在LLM Provider大模型提供者 这是“Provider Neutral”的核心。Runlet不关心你是通过HTTP调用OpenAI还是通过gRPC连接本地模型。它只要求你实现一个简单的接口class LLMProvider: async def generate(self, messages: List[Dict], **kwargs) - str: 接收历史消息列表返回模型生成的文本。 pass你可以为OpenAI、Anthropic、Cohere、本地Llama.cpp、Ollama、甚至是多个模型的组合用于降级或路由实现这个接口。当Agent进入THINKING状态时Runtime就会调用你注入的这个LLMProvider实例。注意这里我选择了异步接口async def。虽然Runlet核心状态机可以是同步的但工具调用和LLM调用通常是I/O密集型操作使用异步可以更高效地利用资源。你可以选择用asyncio.run来驱动一个同步外壳或者让整个Runtime都是异步的。Tool工具 工具是Agent与外界交互的手段。Runlet对工具的定义也非常直接class Tool: name: str description: str parameters: Dict # 描述参数可用JSON Schema async def execute(self, **kwargs) - str: 执行工具返回字符串格式的结果。 pass当LLM决定要调用某个工具时Runlet会根据工具名找到对应的Tool实例传入参数并执行。工具可以是任何东西一个计算器函数、一个网络请求、一段数据库查询代码。Memory记忆 记忆负责存储和管理对话历史。它可以是简单的内存列表也可以是Redis、数据库等外部存储。class Memory: async def add(self, message: Dict): 添加一条消息到历史。 pass async def get_context(self, limit: int 10) - List[Dict]: 获取最近的若干条消息作为上下文。 pass async def clear(self): 清空历史。 passRuntime在每次需要调用LLM前会从Memory中获取上下文消息。这种设计让你可以轻松实现不同长度的上下文窗口、总结式记忆等高级功能。2.3 运行循环一次执行的解剖有了这些组件一次典型的Agent运行循环run方法逻辑如下初始化将状态设为THINKING从Memory中获取对话上下文。思考循环 a.调用LLM将上下文可能加上系统提示词和工具描述发送给LLMProvider。 b.解析决策LLM的回复期望是一个结构化数据如JSON指明是tool_call还是final_answer。 c.执行工具如果是tool_call状态转为ACTING调用对应的Tool然后将结果作为一条新消息加入Memory状态转为OBSERVING并跳回步骤a继续思考。 d.返回结果如果是final_answer状态转为FINISHED将答案加入Memory并返回。这个循环清晰地将LLM的“思考”和工具的“执行”解耦每一步都通过Memory来传递信息。整个Runtime的代码如果不算错误处理和日志核心循环可能不超过100行。这就是“清晰”的含义。3. 实现“Provider Neutral”从接口到具体实践设计理念是美好的但如何落地关键在于我们如何实现和组装这些“Provider”。让我们以连接不同大模型为例看看Runlet如何保持中立。3.1 构建一个统一的LLM Provider适配层我们不会在Runlet的核心代码里写任何import openai。相反我们会创建独立的适配器模块。例如创建一个providers目录runlet/ ├── runtime.py # 核心运行时状态机 ├── memory.py # 记忆抽象 ├── tool.py # 工具抽象 └── providers/ # 各种提供商实现 ├── __init__.py ├── base.py # 基础LLMProvider接口 ├── openai.py # OpenAI适配器 ├── anthropic.py # Anthropic适配器 ├── ollama.py # Ollama适配器 └── qwen.py # 通义千问适配器以OpenAI Provider为例它的实现可能长这样# providers/openai.py import openai from .base import LLMProvider class OpenAIProvider(LLMProvider): def __init__(self, api_key: str, model: str gpt-3.5-turbo, base_url: str None): self.client openai.OpenAI(api_keyapi_key, base_urlbase_url) self.model model async def generate(self, messages, **kwargs): # 注意openai库的新版本默认返回同步客户端这里需要异步调用 # 实际中可以使用await asyncio.to_thread(...)包装同步调用 response await asyncio.to_thread( self.client.chat.completions.create, modelself.model, messagesmessages, **kwargs ) return response.choices[0].message.content而对于本地部署的Qwen模型假设通过类似OpenAI的API接口提供服务实现几乎一样# providers/qwen.py import openai # 仍然使用openai库因为Qwen的API可能兼容OpenAI格式 from .base import LLMProvider class QwenProvider(LLMProvider): def __init__(self, base_url: str http://localhost:8000/v1, model: str qwen): # 将base_url指向本地Qwen服务端点 self.client openai.OpenAI(api_keynot-needed, base_urlbase_url) self.model model async def generate(self, messages, **kwargs): response await asyncio.to_thread( self.client.chat.completions.create, modelself.model, messagesmessages, **kwargs ) return response.choices[0].message.content你看对于Runtime核心来说OpenAIProvider和QwenProvider没有任何区别它们都实现了generate方法。用户在使用时只需要在初始化Runtime时注入不同的Provider实例即可。这完美解决了热词中提到的cursor使用本地模型qwen provider returned error问题——错误被隔离在Provider内部Runtime本身是稳定的。3.2 处理不同模型的输出格式与提示工程“Provider Neutral”的挑战不仅在于调用还在于输入输出。不同模型对消息格式、停止令牌的偏好可能不同。Runlet的策略是将格式化逻辑推给Provider。我们可以在基础LLMProvider接口上增加一个可选的方法class LLMProvider: async def format_messages(self, system_prompt: str, history: List[Dict], tools: List[Tool] None) - List[Dict]: 将系统提示、历史消息和工具描述格式化为该Provider需要的消息列表。 # 默认实现适用于OpenAI格式 messages [{role: system, content: system_prompt}] messages.extend(history) if tools: # 这里可以将工具描述以特定格式如Function Calling附加到消息中 pass return messages这样对于需要特殊消息格式的模型比如Claude你可以在对应的Provider子类中重写这个方法。Runtime只需要调用provider.format_messages(...)来获取最终要发送的消息列表。同理对于解析模型返回的“工具调用”指令Runtime可以期望一个统一的JSON结构但具体的解析和验证逻辑也可以由Provider承担一部分责任或者通过一个可插拔的OutputParser组件来处理。核心原则依然是Runtime定义协议Provider负责实现与特定后端的适配。4. 实战用Runlet构建一个天气预报Agent理论说再多不如动手写一遍。让我们用Runlet构建一个简单的天气预报Agent它可以使用一个“搜索网络”的工具来获取天气信息。我们将使用本地启动的Ollama运行Llama 3模型作为LLM Provider来演示完全的本地化、无云依赖的Agent运行。4.1 定义工具模拟网络搜索首先我们实现一个简单的工具。在真实场景中你可能会调用Serper或Google Search API这里我们模拟一下。# my_tools.py import random from runlet.tool import Tool class WeatherSearchTool(Tool): def __init__(self): self.name get_weather self.description 获取指定城市的当前天气情况。 self.parameters { type: object, properties: { city: {type: string, description: 城市名称例如北京、上海} }, required: [city] } async def execute(self, city: str) - str: # 模拟一个网络请求的延迟 await asyncio.sleep(0.5) # 模拟返回天气数据 weather_conditions [晴, 多云, 阴, 小雨, 中雨, 大雪] temperature random.randint(-5, 35) return f{city}的天气是{random.choice(weather_conditions)}气温{temperature}摄氏度。4.2 实现Ollama Provider假设你已经在本地运行了Ollama并拉取了llama3:8b模型。# my_providers.py import aiohttp import json from runlet.providers.base import LLMProvider class OllamaProvider(LLMProvider): def __init__(self, base_url: str http://localhost:11434, model: str llama3:8b): self.base_url base_url.rstrip(/) self.model model async def generate(self, messages: List[Dict], **kwargs) - str: # Ollama的API格式与OpenAI略有不同 # 我们需要将消息历史转换成Ollama的格式 prompt self._format_messages_to_prompt(messages) async with aiohttp.ClientSession() as session: payload { model: self.model, prompt: prompt, stream: False, # 我们一次性获取完整回复 options: {temperature: 0.7} } async with session.post(f{self.base_url}/api/generate, jsonpayload) as resp: result await resp.json() return result.get(response, ).strip() def _format_messages_to_prompt(self, messages: List[Dict]) - str: # 这是一个简化的转换实际应用可能需要更精细的提示词工程 prompt_parts [] for msg in messages: role msg[role] content msg[content] if role system: prompt_parts.append(fSystem: {content}) elif role user: prompt_parts.append(fUser: {content}) elif role assistant: prompt_parts.append(fAssistant: {content}) prompt_parts.append(Assistant:) # 引导模型开始回复 return \n.join(prompt_parts)4.3 组装并运行Agent现在我们把所有部件组装起来。# main.py import asyncio from runlet import Runtime from runlet.memory import SimpleListMemory from my_tools import WeatherSearchTool from my_providers import OllamaProvider async def main(): # 1. 初始化组件 llm_provider OllamaProvider(modelllama3:8b) memory SimpleListMemory() tools [WeatherSearchTool()] # 2. 创建Runtime agent Runtime( llm_providerllm_provider, memorymemory, toolstools, system_prompt你是一个有帮助的天气助手。你可以使用工具来查询天气。请用中文回复。 ) # 3. 运行Agent user_query 上海今天天气怎么样 print(f用户: {user_query}) final_response await agent.run(user_query) print(f助手: {final_response}) # 查看记忆 print(\n对话历史:) for msg in memory.messages: print(f{msg[role]}: {msg[content][:100]}...) if __name__ __main__: asyncio.run(main())当你运行这段代码时Runtime会将用户问题“上海今天天气怎么样”加入记忆。调用OllamaProvider将系统提示、历史当前只有用户问题和工具描述发送给本地Llama 3模型。Llama 3模型如果提示词引导得当会识别出需要调用get_weather工具并返回一个结构化的工具调用请求。Runtime解析这个请求调用WeatherSearchTool.execute(city上海)。工具返回模拟的天气结果Runtime将其作为一条新消息role: “tool”, content: “上海...”加入记忆。Runtime再次调用LLM这次的历史包含了工具执行结果LLM会生成一个总结性的自然语言回复例如“根据查询上海今天是晴天气温22摄氏度。”Runtime将此回复作为最终答案返回并结束循环。整个过程完全在本地进行没有调用任何外部云服务。你可以通过更换llm_provider为OpenAIProvider并填入API密钥瞬间切换到GPT-4而不需要修改Runtime、Tool或Memory的任何一行代码。这就是“Provider Neutral”带来的威力。5. 深入细节错误处理、流式输出与性能考量一个健壮的Runtime不能只处理Happy Path。让我们深入几个关键细节看看Runlet如何保持“小而清晰”的同时具备实用性。5.1 健壮的错误处理与状态回退在状态机运行中任何一步都可能出错LLM调用可能超时provider rejected a test request工具执行可能异常LLM的输出可能无法解析。Runlet的核心设计原则是错误不应导致Runtime崩溃而应被捕获并转换为可管理的状态。我们在Runtime的run循环中需要加入全面的try-catch# 在 run 方法的主循环中 try: if self.state State.THINKING: llm_response await self.llm_provider.generate(context_messages) parsed_action self._parse_llm_response(llm_response) # 可能抛出解析异常 elif self.state State.ACTING: tool_result await self.current_tool.execute(**self.current_tool_args) # 可能抛出执行异常 except (LLMProviderError, ToolExecutionError, ParsingError) as e: # 捕获特定异常 await self.memory.add({ role: system, content: f执行过程中发生错误{str(e)}。请调整你的策略。 }) # 状态回退到THINKING让LLM根据错误信息决定下一步如重试、换工具或向用户求助 self.state State.THINKING continue # 继续循环 except Exception as e: # 未预料的异常将状态设为ERROR并向上抛出或记录日志 self.state State.ERROR raise RuntimeError(fAgent运行时发生未知错误: {e}) from e此外对于网络调用LLM、工具必须设置合理的超时和重试逻辑。这些逻辑最好封装在各自的Provider内部。例如在OllamaProvider.generate中可以使用aiohttp的超时设置和重试装饰器。5.2 支持流式输出Streaming现代LLM应用体验离不开流式输出。Runlet作为Runtime应该支持将LLM的流式响应实时传递给上层。我们可以通过回调函数Callback或异步生成器Async Generator来实现。一种简洁的设计是在LLMProvider.generate方法中增加一个stream_callback参数class LLMProvider: async def generate(self, messages: List[Dict], stream_callback: Optional[Callable[[str], None]] None, **kwargs) - str: 如果提供了stream_callback则每产生一个token就调用一次。 pass在OpenAIProvider中的实现async def generate(self, messages, stream_callbackNone, **kwargs): if stream_callback: # 流式模式 stream await asyncio.to_thread( self.client.chat.completions.create, modelself.model, messagesmessages, streamTrue, **kwargs ) full_content async for chunk in stream: delta chunk.choices[0].delta.content if delta is not None: full_content delta stream_callback(delta) # 实时回调 return full_content else: # 非流式模式原有逻辑 ...在Runtime中如果用户传入了流式回调则在调用LLM时传入。这样上层应用如Web服务器就可以将token实时推送到前端实现打字机效果。Runlet核心只负责传递不关心数据如何被消费。5.3 内存管理与性能优化“小”也意味着对资源敏感。Runlet的默认SimpleListMemory只是将消息存在内存的列表里对于长对话这会导致上下文越来越长最终可能超过模型的令牌限制。解决方案是引入记忆窗口或记忆总结。这可以通过实现更复杂的Memory类来完成例如WindowedMemory只保留最近N条消息。SummarizingMemory当消息超过一定数量时调用LLM可以是一个更小的、便宜的模型对旧消息进行总结将总结文本作为一条新消息存入并丢弃原始旧消息。这些高级Memory实现可以完全独立于Runtime核心通过依赖注入来使用。这再次体现了清晰抽象的威力——核心逻辑不变只需换一个“记忆提供者”就能获得全新的能力。另一个性能考量是工具执行的并行化。如果一个Agent计划同时调用多个不依赖的工具理论上可以并行执行。这可以通过在Runtime中引入一个ToolExecutor组件来实现它管理一个任务池。但需要注意的是这增加了状态管理的复杂性需要等待所有并行工具完成才能进入下一步。对于“小而清晰”的初始目标Runlet可以暂不支持并行工具调用保持状态机的线性与简单。如果需要可以作为扩展功能提供。6. 对比与定位Runlet在生态中的位置在AI Agent开发领域已经有很多优秀的框架和库。理解Runlet与它们的区别能更好地定位它的使用场景。特性/框架LangChain / LlamaIndexAutoGenSemantic KernelRunlet (本文构想)设计哲学“全家桶”式框架提供大量预制组件、工具链和集成。专注于多Agent协作与对话提供复杂的对话模式。微软出品深度集成.NET生态强调“技能”与“插件”。“极简内核”只做最核心的运行时调度其他皆可插拔。核心价值快速构建复杂应用开箱即用社区资源丰富。便捷地构建多Agent对话系统。在微软技术栈内构建AI应用与Copilot Studio等结合好。极致灵活与透明无供应商锁定易于理解、调试和定制。依赖复杂度高。依赖众多安装包体积大。中高。中取决于使用的部分。极低。核心只有Python标准库和几个接口定义。Provider绑定较强。虽然支持多种模型但内部实现与OpenAI格式耦合较深切换成本存在。支持多种模型但框架本身有一定复杂性。支持多种模型但作为微软框架与Azure AI服务集成最顺滑。完全中立。核心代码零导入任何SDK绑定由用户决定。适用场景需要快速原型验证、利用丰富生态的中大型应用。研究或构建需要多个Agent相互对话、辩论、协作的系统。企业内基于.NET技术栈开发AI功能尤其是与微软产品集成。资源受限环境边缘设备、对依赖和性能有严苛要求、需要深度定制Agent逻辑、作为教学工具理解Agent原理。学习曲线中高。需要理解其众多的抽象概念Chains, Agents, Tools, Memory。中。需要理解其Agent和GroupChat的概念。中。需要理解其Kernel, Skills, Plugins的概念。低。核心就是一个状态机循环代码即文档。Runlet的定位非常明确它不是要取代这些框架而是填补一个细分空白。当你需要将一个智能助手嵌入到一个小型IoT设备树莓派中。在Serverless函数对冷启动时间和包大小敏感中运行一个简单的Agent逻辑。作为一个教学示例向新手清晰地展示Agent是如何一步步“思考-行动-观察”循环的。在一个已有复杂系统中需要引入AI能力但希望最小化外部依赖和框架侵入。在这些场景下Runlet的“小而清晰、Provider Neutral”特性就成为了巨大的优势。你可以把它看作是一个乐高积木的“底板”其他所有功能记忆、工具、模型都是你可以自由选择并插上去的“积木块”。7. 扩展思路将Runlet变得“可用”目前我们讨论的Runlet还是一个概念和简单的实现。要让它成为一个真正“可用”的库还需要考虑很多工程化细节。7.1 配置化与工厂模式用户不应该每次都手动实例化Provider、Tool和Memory。我们可以提供一个简单的配置系统。# config.yaml runtime: system_prompt: 你是一个有帮助的助手。 max_turns: 10 llm: provider: openai # 或 ollama, anthropic, qwen model: gpt-3.5-turbo api_key: ${OPENAI_API_KEY} # 支持环境变量 base_url: null memory: type: simple_list # 或 redis, postgres max_messages: 20 tools: - name: search_web class: my_tools.WebSearchTool params: api_key: ${SERPER_API_KEY} - name: calculate class: my_tools.CalculatorTool然后一个AgentFactory可以根据这个配置文件动态导入对应的类并创建出配置好的Runtime实例。这大大降低了使用门槛。7.2 可观测性与日志对于调试和生产监控Runtime需要提供详细的日志。我们可以在状态机的每个关键节点状态转换、调用LLM开始/结束、调用工具开始/结束触发日志事件。这些事件可以通过Python的logging模块输出也可以被一个可插拔的EventHook系统捕获以便集成到APM应用性能监控系统中。class Runtime: def __init__(self, ..., event_hooks: List[EventHook] None): self.event_hooks event_hooks or [] async def _trigger_event(self, event_name: str, data: Dict): for hook in self.event_hooks: await hook.on_event(event_name, data) # 在状态转换、LLM调用等处调用 _trigger_event7.3 测试策略如何测试一个Agent Runtime我们可以采用分层测试单元测试单独测试LLMProvider、Tool、Memory的实现。对于LLM Provider可以使用unittest.mock来模拟HTTP响应。集成测试测试整个Runtime的组装。这里可以使用一个Mock LLM Provider它不调用真实模型而是根据输入消息返回预设的、符合格式的响应。这可以验证状态机在各种预设路径如直接回答、调用工具一次后回答、调用工具多次下是否能正确运转。端到端测试在测试环境中使用一个轻量级、确定性的真实模型例如一个非常小的本地模型来运行几个关键用例确保整个链路在真实环境下也能工作。Mock LLM Provider是测试中的利器它让你可以在不依赖不稳定且昂贵的真实LLM API的情况下全面测试你的Agent逻辑和工具集成。构建Runlet这样的项目更像是在打造一个精密的瑞士军刀底座而不是一个功能齐全的厨房。它可能不会满足你所有的需求但它会给你最需要的控制权、清晰度和自由。在AI应用开发日益复杂的今天有时回归简单恰恰是通往更强大、更可靠系统的捷径。