FlowLLM：让基于LLM的HTTP/MCP服务开发更简单
_{如果觉得有用，欢迎给个 ⭐ Star，您的支持是我们持续改进的动力}

English | 简体中文

---

📖 简介

FlowLLM 将 LLM/Embedding/vector_store 能力封装为 HTTP/MCP 服务，适用于 AI 对话助手、RAG 应用、工作流服务等场景，并可集成到支持 MCP 的客户端工具中。

🏗️ 架构概览

🌟 基于FlowLLM的应用

项目名	描述
ReMe	面向智能体的记忆管理工具包

📢 最近更新

日期	更新内容
2025-11-15	新增 File Tool Op 功能，提供 13 个文件操作工具，支持文件读取、写入、编辑、搜索、目录操作、系统命令执行和任务管理等功能
2025-11-14	新增 Token 计数能力，支持通过 self.token_count() 方法准确计算消息和工具的 token 数量，支持多种后端（base、openai、hf），配置示例参考 default.yaml

📚 学习资料分享

项目开发者会在这里分享最近的学习资料。

日期	标题	描述
2025-11-14	HaluMem解读	HaluMem: Evaluating Hallucinations in Memory Systems of Agents 解读
2025-11-13	Gemini CLI 上下文管理机制	Gemini CLI 的多层上下文管理策略
2025-11-10	上下文管理指南	上下文管理指南
2025-11-10	LangChain&Manus视频资料	LangChain & Manus Context Management Video

⭐ 核心特性

• 简单易用的 Op 开发：继承 BaseOp 或 BaseAsyncOp 基类，实现业务逻辑即可。FlowLLM提供了延迟初始化的 LLM、Embedding 模型和向量库，开发者只需通过 self.llm、self.embedding_model、self.vector_store 即可轻松使用这些资源。同时FlowLLM提供了完整的 Prompt 模板管理能力，通过 prompt_format() 和 get_prompt() 方法进行格式化和使用。此外，FlowLLM 还内置了 Token 计数能力，通过 self.token_count() 方法可以准确计算消息和工具的 token 数量，支持多种后端（base、openai、hf 等）。

• 灵活的 Flow 编排：通过 YAML 配置文件将 Op 组合成 Flow，支持灵活的编排方式。>> 表示串行组合，| 表示并行组合，例如 SearchOp() >> (AnalyzeOp() | TranslateOp()) >> FormatOp() 可构建复杂的工作流。定义输入输出 Schema 后，使用 flowllm config=your_config 命令即可启动服务。

• 自动生成服务：配置完成后，FlowLLM 会自动生成 HTTP、MCP 和 CMD 服务。HTTP 服务提供标准的 RESTFul API，支持同步 JSON 响应和 HTTP Stream 流式响应。MCP 服务会自动注册为 Model Context Protocol 工具，可集成到支持 MCP 的客户端中。CMD 服务支持命令行模式执行单个 Op，适合快速测试和调试。

---

⚡ 快速开始

📦 Step0 安装

📥 From PyPI

pip install flowllm

🔧 From Source

git clone https://github.com/flowllm-ai/flowllm.git
cd flowllm
pip install -e .

详细安装与配置方法请参考 安装指南。

⚙️ 配置

创建 .env 文件，配置 API Key。你可以从 example.env 复制并修改：

cp example.env .env

然后在 .env 文件中配置你的 API Key：

FLOW_LLM_API_KEY=sk-xxxx
FLOW_LLM_BASE_URL=https://xxxx/v1
FLOW_EMBEDDING_API_KEY=sk-xxxx
FLOW_EMBEDDING_BASE_URL=https://xxxx/v1

详细配置说明请参考 配置指南。

🛠️ Step1 构建Op

from flowllm.core.context import C
from flowllm.core.op import BaseAsyncOp
from flowllm.core.schema import Message
from flowllm.core.enumeration import Role

@C.register_op()
class SimpleChatOp(BaseAsyncOp):
    async def async_execute(self):
        query = self.context.get("query", "")
        messages = [Message(role=Role.USER, content=query)]

        # 使用 token_count 方法计算 token 数量
        token_num = self.token_count(messages)
        print(f"Input tokens: {token_num}")

        response = await self.llm.achat(messages=messages)
        self.context.response.answer = response.content.strip()

详细内容请参考 简单 Op 指南、LLM Op 指南 和 高级 Op 指南（包含 Embedding、VectorStore 和并发执行等高级功能）。

📝 Step2 配置config

以下示例展示如何构建一个 MCP（Model Context Protocol）服务。创建配置文件 my_mcp_config.yaml：

backend: mcp

mcp:
  transport: sse
  host: "0.0.0.0"
  port: 8001

flow:
  demo_mcp_flow:
    flow_content: MockSearchOp()
    description: "Search results for a given query."
    input_schema:
      query:
        type: string
        description: "User query"
        required: true

llm:
  default:
    backend: openai_compatible
    model_name: qwen3-30b-a3b-instruct-2507
    params:
      temperature: 0.6
    token_count: # 可选，配置 token 计数后端
      model_name: Qwen/Qwen3-30B-A3B-Instruct-2507
      backend: hf  # 支持 base、openai、hf 等
      params:
        use_mirror: true

🚀 Step3 启动 MCP 服务

flowllm \
  config=my_mcp_config \
  backend=mcp \  # 可选，覆盖config配置
  mcp.transport=sse \  # 可选，覆盖config配置
  mcp.port=8001 \  # 可选，覆盖config配置
  llm.default.model_name=qwen3-30b-a3b-thinking-2507  # 可选，覆盖config配置

服务启动后可以参考Client Guide来使用服务，可以直接获取模型所需要的tool_call。

---

📚 详细文档

🚀 入门指南

• 安装指南
• 配置指南

🔧 Op 开发

• Op 介绍
• 简单 Op 指南
• LLM Op 指南
• 高级 Op 指南
• Tool Op 指南
• File Tool Op 指南
• Vector Store 指南

🔀 Flow 编排

• Flow 指南

🌐 服务使用

• HTTP 服务指南
• HTTP Stream 指南
• MCP 服务指南
• CMD 服务指南
• 客户端指南

---

🤝 参与贡献

欢迎各种形式的贡献！具体参与方式请参考 贡献指南。

---

📄 许可证

本项目采用 Apache 2.0 许可证。

---

Star History

---

GitHub • 文档 • PyPI