在AI应用开发中我们经常遇到一个核心痛点通用大模型虽然知识渊博但无法准确回答特定领域的问题。比如询问公司内部产品信息、技术文档细节或私有数据时模型往往给出笼统甚至错误的答案。这正是RAG检索增强生成技术要解决的关键问题。本文将手把手带你搭建完整的私有知识库系统从基础概念到实战部署涵盖文档解析、向量检索、排序优化等核心环节。无论你是AI初学者还是有经验的开发者都能通过本文学会构建企业级知识库应用。1. RAG技术核心概念解析1.1 什么是RAG技术RAGRetrieval-Augmented Generation检索增强生成是一种将信息检索与大模型生成相结合的技术框架。其核心思想是在回答用户问题前先从知识库中检索相关文档片段然后将这些片段作为上下文提供给大模型从而生成更准确、更相关的回答。传统大模型依赖训练时学到的知识存在知识截止日期和幻觉问题。RAG通过实时检索外部知识源让模型能够访问最新、最相关的信息特别适合企业知识管理、客服系统、文档问答等场景。1.2 RAG与传统微调的对比很多开发者会困惑到底是选择RAG还是模型微调这两种技术各有适用场景RAG的优势知识更新成本低只需更新知识库文档无需重新训练模型可解释性强可以追溯答案来源显示参考文档处理长尾知识能够覆盖训练数据中不存在的特定领域知识避免灾难性遗忘不会影响模型原有的通用能力微调的适用场景需要改变模型推理风格或输出格式领域术语和表达方式与通用语言差异很大任务需要特定的思维链或推理模式在实际项目中RAG和微调经常结合使用先用微调优化模型的基础能力再用RAG提供实时知识支持。1.3 RAG系统核心组件一个完整的RAG系统包含以下关键组件文档加载与解析支持PDF、Word、Excel、HTML等多种格式文档的解析文本切分Chunking将长文档切分为适合检索的文本片段向量化Embedding将文本转换为数值向量用于相似度计算向量数据库存储和快速检索向量数据检索器Retriever根据查询找到最相关的文档片段重排序Reranker对初步检索结果进行精细排序生成模型LLM基于检索到的上下文生成最终答案2. 环境准备与工具选型2.1 基础环境要求搭建RAG系统前需要准备以下环境硬件要求CPU4核以上建议8核内存16GB以上处理大量文档时建议32GB存储SSD硬盘至少100GB可用空间GPU可选用于加速Embedding和推理RTX 3080以上软件环境操作系统LinuxUbuntu 20.04或Windows 10Python3.8-3.11版本包管理Conda或pip2.2 核心工具链选择向量数据库选项Chroma轻量级适合入门和开发测试FAISSFacebook开源性能优秀适合生产环境Qdrant云原生设计支持分布式部署Pinecone全托管服务免运维Embedding模型选择text-embedding-3-smallOpenAI出品平衡性能与成本BAAI/bge-small-zh中文优化开源免费moka-ai/m3e-base中文任务表现优秀大模型选择GPT系列API调用效果稳定但需要网络Qwen系列阿里开源中文支持优秀ChatGLM系列清华开源中英双语能力均衡2.3 项目结构规划建议采用以下目录结构组织RAG项目rag-project/ ├── data/ # 原始文档存储 │ ├── pdfs/ # PDF文档 │ ├── docs/ # Word文档 │ └── excels/ # Excel文件 ├── processed/ # 处理后的数据 │ ├── chunks/ # 文本切片 │ └── vectors/ # 向量数据 ├── src/ # 源代码 │ ├── document_loader.py # 文档加载器 │ ├── chunking.py # 文本切分 │ ├── embedding.py # 向量化 │ ├── retriever.py # 检索器 │ └── generator.py # 答案生成 ├── config/ # 配置文件 │ └── settings.yaml # 系统配置 └── tests/ # 测试文件3. 文档处理与向量化实战3.1 文档加载与解析文档解析是RAG系统的第一步需要处理多种格式的文档。以下是完整的文档加载实现# document_loader.py import os from typing import List, Dict, Any import PyPDF2 from docx import Document import pandas as pd from langchain.document_loaders import ( PDFPlumberLoader, UnstructuredWordDocumentLoader, UnstructuredFileLoader ) class DocumentProcessor: def __init__(self, supported_formats: List[str] None): self.supported_formats supported_formats or [.pdf, .docx, .txt, .md, .xlsx] def load_document(self, file_path: str) - List[Dict[str, Any]]: 加载单个文档并返回结构化数据 if not os.path.exists(file_path): raise FileNotFoundError(f文档不存在: {file_path}) file_ext os.path.splitext(file_path)[1].lower() if file_ext .pdf: return self._load_pdf(file_path) elif file_ext .docx: return self._load_docx(file_path) elif file_ext in [.xlsx, .xls]: return self._load_excel(file_path) elif file_ext in [.txt, .md]: return self._load_text(file_path) else: raise ValueError(f不支持的文档格式: {file_ext}) def _load_pdf(self, file_path: str) - List[Dict[str, Any]]: 加载PDF文档 documents [] try: loader PDFPlumberLoader(file_path) pages loader.load() for i, page in enumerate(pages): documents.append({ content: page.page_content, metadata: { source: file_path, page: i 1, type: pdf } }) except Exception as e: print(fPDF解析错误 {file_path}: {e}) return documents def _load_docx(self, file_path: str) - List[Dict[str, Any]]: 加载Word文档 documents [] try: doc Document(file_path) full_text [] for paragraph in doc.paragraphs: if paragraph.text.strip(): full_text.append(paragraph.text) content \n.join(full_text) documents.append({ content: content, metadata: { source: file_path, type: docx } }) except Exception as e: print(fWord文档解析错误 {file_path}: {e}) return documents # 使用示例 if __name__ __main__: processor DocumentProcessor() documents processor.load_document(example.pdf) print(f加载了 {len(documents)} 个文档片段)3.2 智能文本切分策略文本切分是影响检索效果的关键因素。以下实现多种切分策略# chunking.py import re from typing import List, Dict, Any from langchain.text_splitter import RecursiveCharacterTextSplitter, MarkdownTextSplitter class SmartChunker: def __init__(self, chunk_size: int 1000, chunk_overlap: int 200): self.chunk_size chunk_size self.chunk_overlap chunk_overlap self.text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , 、, ] ) def chunk_documents(self, documents: List[Dict[str, Any]]) - List[Dict[str, Any]]: 对文档进行智能切分 chunks [] for doc in documents: content doc[content] metadata doc[metadata] # 根据文档类型选择切分策略 if metadata.get(type) md: text_chunks self._chunk_markdown(content) else: text_chunks self._chunk_general(content) # 为每个切片添加元数据 for i, chunk_text in enumerate(text_chunks): chunk_metadata metadata.copy() chunk_metadata[chunk_id] i chunk_metadata[chunk_size] len(chunk_text) chunks.append({ content: chunk_text, metadata: chunk_metadata }) return chunks def _chunk_general(self, text: str) - List[str]: 通用文本切分 return self.text_splitter.split_text(text) def _chunk_markdown(self, text: str) - List[str]: Markdown文档切分保持结构完整性 # 按标题切分 pattern r(# .?)(?# |\Z) sections re.findall(pattern, text, re.DOTALL) chunks [] for section in sections: if len(section) self.chunk_size: chunks.append(section) else: # 如果章节过长进一步切分 sub_chunks self.text_splitter.split_text(section) chunks.extend(sub_chunks) return chunks def chunk_by_semantic(self, text: str, max_tokens: int 512) - List[str]: 基于语义的智能切分 sentences re.split(r[。\.!?;], text) sentences [s.strip() for s in sentences if s.strip()] chunks [] current_chunk for sentence in sentences: # 估算token数量中文大致按字符数计算 if len(current_chunk) len(sentence) max_tokens * 2: # 粗略估算 current_chunk sentence 。 else: if current_chunk: chunks.append(current_chunk.strip()) current_chunk sentence 。 if current_chunk: chunks.append(current_chunk.strip()) return chunks3.3 向量化与存储向量化是将文本转换为数值表示的关键步骤# embedding.py import numpy as np from typing import List, Dict, Any import requests from sentence_transformers import SentenceTransformer class EmbeddingEngine: def __init__(self, model_name: str BAAI/bge-small-zh, device: str cpu): self.model SentenceTransformer(model_name, devicedevice) self.model_name model_name def encode_texts(self, texts: List[str]) - np.ndarray: 将文本列表转换为向量 embeddings self.model.encode(texts, normalize_embeddingsTrue) return embeddings def encode_single(self, text: str) - np.ndarray: 将单个文本转换为向量 return self.encode_texts([text])[0] def batch_encode(self, chunks: List[Dict[str, Any]], batch_size: int 32) - List[Dict[str, Any]]: 批量处理文本切片并添加向量 texts [chunk[content] for chunk in chunks] # 分批处理避免内存溢出 all_embeddings [] for i in range(0, len(texts), batch_size): batch_texts texts[i:i batch_size] batch_embeddings self.encode_texts(batch_texts) all_embeddings.extend(batch_embeddings) # 将向量添加到切片数据中 for i, chunk in enumerate(chunks): chunk[embedding] all_embeddings[i].tolist() return chunks # 向量存储实现 class VectorStore: def __init__(self, persist_directory: str ./vector_store): self.persist_directory persist_directory self.vectors [] self.metadata [] os.makedirs(persist_directory, exist_okTrue) def add_vectors(self, chunks: List[Dict[str, Any]]): 添加向量到存储 for chunk in chunks: if embedding in chunk: self.vectors.append(chunk[embedding]) self.metadata.append(chunk[metadata]) def similarity_search(self, query_embedding: List[float], top_k: int 5) - List[Dict[str, Any]]: 相似度搜索 if not self.vectors: return [] # 计算余弦相似度 query_vec np.array(query_embedding) vectors np.array(self.vectors) # 归一化 query_vec query_vec / np.linalg.norm(query_vec) vectors vectors / np.linalg.norm(vectors, axis1, keepdimsTrue) # 计算相似度 similarities np.dot(vectors, query_vec) # 获取最相似的top_k个结果 top_indices np.argsort(similarities)[-top_k:][::-1] results [] for idx in top_indices: results.append({ content: self.metadata[idx].get(content, ), metadata: self.metadata[idx], similarity: float(similarities[idx]) }) return results def save(self): 保存向量数据 data { vectors: self.vectors, metadata: self.metadata } with open(os.path.join(self.persist_directory, vectors.pkl), wb) as f: pickle.dump(data, f) def load(self): 加载向量数据 path os.path.join(self.persist_directory, vectors.pkl) if os.path.exists(path): with open(path, rb) as f: data pickle.load(f) self.vectors data[vectors] self.metadata data[metadata]4. 检索与重排序优化4.1 多路检索策略单一检索方式可能无法覆盖所有相关文档实现多路检索提升召回率# retriever.py from typing import List, Dict, Any import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer from rank_bm25 import BM25Okapi import jieba class HybridRetriever: def __init__(self, vector_store: VectorStore): self.vector_store vector_store self.bm25_index None self.tfidf_vectorizer None self.documents [] def build_keyword_index(self, chunks: List[Dict[str, Any]]): 构建关键词检索索引 self.documents [chunk[content] for chunk in chunks] # 中文分词 tokenized_docs [list(jieba.cut(doc)) for doc in self.documents] self.bm25_index BM25Okapi(tokenized_docs) # TF-IDF向量化 self.tfidf_vectorizer TfidfVectorizer() self.tfidf_matrix self.tfidf_vectorizer.fit_transform(self.documents) def hybrid_search(self, query: str, top_k: int 10, weights: Dict[str, float] None) - List[Dict[str, Any]]: 混合检索 if weights is None: weights {vector: 0.6, bm25: 0.3, tfidf: 0.1} # 1. 向量检索 embedding_engine EmbeddingEngine() query_embedding embedding_engine.encode_single(query) vector_results self.vector_store.similarity_search(query_embedding, top_k * 3) # 2. BM25检索 tokenized_query list(jieba.cut(query)) bm25_scores self.bm25_index.get_scores(tokenized_query) bm25_indices np.argsort(bm25_scores)[-top_k * 3:][::-1] # 3. TF-IDF检索 query_vec self.tfidf_vectorizer.transform([query]) tfidf_scores (self.tfidf_matrix * query_vec.T).toarray().flatten() tfidf_indices np.argsort(tfidf_scores)[-top_k * 3:][::-1] # 结果融合 fused_results self._fuse_results( vector_results, bm25_indices, tfidf_indices, weights, top_k ) return fused_results def _fuse_results(self, vector_results, bm25_indices, tfidf_indices, weights, top_k): 结果融合算法 # 实现RRFReciprocal Rank Fusion等融合算法 scores {} # 向量检索结果评分 for i, result in enumerate(vector_results): doc_id result[metadata].get(chunk_id, i) scores[doc_id] scores.get(doc_id, 0) weights[vector] * (1 / (i 1)) # BM25结果评分 for i, idx in enumerate(bm25_indices): scores[idx] scores.get(idx, 0) weights[bm25] * (1 / (i 1)) # TF-IDF结果评分 for i, idx in enumerate(tfidf_indices): scores[idx] scores.get(idx, 0) weights[tfidf] * (1 / (i 1)) # 按综合评分排序 sorted_docs sorted(scores.items(), keylambda x: x[1], reverseTrue)[:top_k] final_results [] for doc_id, score in sorted_docs: # 根据doc_id获取文档内容 # 这里需要根据实际存储结构实现 final_results.append({ doc_id: doc_id, score: score, content: self.documents[doc_id] if doc_id len(self.documents) else }) return final_results4.2 重排序优化初步检索结果可能包含相关性不高的文档通过重排序提升精度# reranker.py from typing import List, Dict, Any import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification class Reranker: def __init__(self, model_name: str BAAI/bge-reranker-base): self.tokenizer AutoTokenizer.from_pretrained(model_name) self.model AutoModelForSequenceClassification.from_pretrained(model_name) self.model.eval() def rerank(self, query: str, documents: List[Dict[str, Any]], top_k: int 5) - List[Dict[str, Any]]: 对检索结果进行重排序 if not documents: return [] # 准备输入数据 pairs [(query, doc[content]) for doc in documents] # 批量推理 with torch.no_grad(): inputs self.tokenizer( pairs, paddingTrue, truncationTrue, return_tensorspt, max_length512 ) scores self.model(**inputs).logits[:, 0].cpu().numpy() # 添加重排序分数并排序 for i, doc in enumerate(documents): doc[rerank_score] float(scores[i]) # 按重排序分数排序 reranked_docs sorted(documents, keylambda x: x[rerank_score], reverseTrue) return reranked_docs[:top_k] def cross_encoder_rerank(self, query: str, documents: List[Dict[str, Any]]) - List[Dict[str, Any]]: 使用交叉编码器进行更精细的重排序 # 交叉编码器同时编码query和document效果更好但计算成本更高 pass5. 完整RAG系统集成5.1 系统架构整合将各个组件整合为完整的RAG系统# rag_system.py import os import yaml from typing import List, Dict, Any from document_loader import DocumentProcessor from chunking import SmartChunker from embedding import EmbeddingEngine, VectorStore from retriever import HybridRetriever from reranker import Reranker class RAGSystem: def __init__(self, config_path: str config/settings.yaml): self.config self._load_config(config_path) self.document_processor DocumentProcessor() self.chunker SmartChunker( chunk_sizeself.config.get(chunk_size, 1000), chunk_overlapself.config.get(chunk_overlap, 200) ) self.embedding_engine EmbeddingEngine( model_nameself.config.get(embedding_model, BAAI/bge-small-zh) ) self.vector_store VectorStore( persist_directoryself.config.get(vector_store_path, ./vector_store) ) self.retriever HybridRetriever(self.vector_store) self.reranker Reranker() # 加载现有向量数据 if os.path.exists(self.vector_store.persist_directory): self.vector_store.load() def _load_config(self, config_path: str) - Dict[str, Any]: 加载配置文件 with open(config_path, r, encodingutf-8) as f: return yaml.safe_load(f) def ingest_documents(self, document_paths: List[str]): 文档注入流程 all_chunks [] for doc_path in document_paths: print(f处理文档: {doc_path}) # 1. 加载文档 documents self.document_processor.load_document(doc_path) # 2. 文本切分 chunks self.chunker.chunk_documents(documents) # 3. 向量化 chunks_with_embeddings self.embedding_engine.batch_encode(chunks) all_chunks.extend(chunks_with_embeddings) # 4. 存储向量 self.vector_store.add_vectors(all_chunks) self.vector_store.save() # 5. 构建关键词索引 self.retriever.build_keyword_index(all_chunks) print(f文档注入完成共处理 {len(all_chunks)} 个文本切片) def query(self, question: str, top_k: int 5) - Dict[str, Any]: 查询处理流程 # 1. 混合检索 retrieved_docs self.retriever.hybrid_search(question, top_k * 3) # 2. 重排序 reranked_docs self.reranker.rerank(question, retrieved_docs, top_k) # 3. 准备上下文 context \n\n.join([doc[content] for doc in reranked_docs]) # 4. 生成回答这里需要接入LLM answer self._generate_answer(question, context) return { question: question, answer: answer, source_documents: reranked_docs, context: context } def _generate_answer(self, question: str, context: str) - str: 调用LLM生成答案 # 这里可以接入OpenAI API、本地模型等 prompt f基于以下上下文信息请回答问题。如果上下文不足以回答问题请说明。 上下文 {context} 问题{question} 回答 # 实际项目中这里会调用LLM API # 例如response openai.ChatCompletion.create(...) return 这是基于检索到文档生成的答案示例。实际项目中需要接入真实LLM。 # 配置文件示例 config_content chunk_size: 1000 chunk_overlap: 200 embedding_model: BAAI/bge-small-zh vector_store_path: ./vector_store reranker_model: BAAI/bge-reranker-base max_retrieved_docs: 20 # 创建配置文件 os.makedirs(config, exist_okTrue) with open(config/settings.yaml, w, encodingutf-8) as f: f.write(config_content)5.2 实时API服务将RAG系统封装为API服务# api_server.py from flask import Flask, request, jsonify from rag_system import RAGSystem import logging app Flask(__name__) logging.basicConfig(levellogging.INFO) # 初始化RAG系统 rag_system RAGSystem() app.route(/api/query, methods[POST]) def handle_query(): 处理查询请求 try: data request.get_json() question data.get(question, ) top_k data.get(top_k, 5) if not question: return jsonify({error: 问题不能为空}), 400 result rag_system.query(question, top_k) return jsonify(result) except Exception as e: logging.error(f查询处理错误: {e}) return jsonify({error: 内部服务器错误}), 500 app.route(/api/ingest, methods[POST]) def handle_ingest(): 处理文档注入请求 try: data request.get_json() document_paths data.get(document_paths, []) if not document_paths: return jsonify({error: 文档路径列表不能为空}), 400 rag_system.ingest_documents(document_paths) return jsonify({message: 文档注入成功}) except Exception as e: logging.error(f文档注入错误: {e}) return jsonify({error: 文档注入失败}), 500 app.route(/api/health, methods[GET]) def health_check(): 健康检查 return jsonify({status: healthy}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)6. 性能优化与生产部署6.1 检索性能优化索引优化策略# optimization.py import faiss import numpy as np from typing import List class FAISSIndexOptimizer: def __init__(self, dimension: int 384): self.dimension dimension self.index None def create_index(self, vectors: List[List[float]], index_type: str IVF): 创建优化的FAISS索引 vectors_array np.array(vectors).astype(float32) if index_type IVF: # 倒排文件索引适合大规模数据 nlist 100 # 聚类中心数量 quantizer faiss.IndexFlatIP(self.dimension) self.index faiss.IndexIVFFlat(quantizer, self.dimension, nlist) # 训练索引 self.index.train(vectors_array) self.index.add(vectors_array) elif index_type HNSW: # 基于图的索引适合高召回率场景 self.index faiss.IndexHNSWFlat(self.dimension, 32) self.index.add(vectors_array) return self.index def optimize_search_params(self, nprobe: int 10): 优化搜索参数 if hasattr(self.index, nprobe): self.index.nprobe nprobe # 搜索的聚类中心数量 # 批量处理优化 class BatchProcessor: def __init__(self, batch_size: int 100): self.batch_size batch_size def process_large_dataset(self, documents: List[str], processing_func): 分批处理大规模数据集 results [] for i in range(0, len(documents), self.batch_size): batch documents[i:i self.batch_size] batch_results processing_func(batch) results.extend(batch_results) return results6.2 缓存策略实现多级缓存优化# cache.py import redis import pickle from typing import Any, Optional import hashlib class CacheManager: def __init__(self, redis_url: str redis://localhost:6379/0): self.redis_client redis.from_url(redis_url) self.local_cache {} # 本地内存缓存 self.ttl 3600 # 缓存过期时间秒 def _generate_key(self, query: str) - str: 生成缓存键 return hashlib.md5(query.encode()).hexdigest() def get(self, query: str) - Optional[Any]: 获取缓存结果 # 首先检查本地缓存 cache_key self._generate_key(query) if cache_key in self.local_cache: return self.local_cache[cache_key] # 检查Redis缓存 redis_data self.redis_client.get(cache_key) if redis_data: result pickle.loads(redis_data) # 更新本地缓存 self.local_cache[cache_key] result return result return None def set(self, query: str, result: Any): 设置缓存 cache_key self._generate_key(query) # 更新本地缓存 self.local_cache[cache_key] result # 更新Redis缓存 redis_data pickle.dumps(result) self.redis_client.setex(cache_key, self.ttl, redis_data) def clear(self, pattern: str *): 清除缓存 keys self.redis_client.keys(pattern) if keys: self.redis_client.delete(*keys) self.local_cache.clear()7. 常见问题与解决方案7.1 检索效果不佳问题排查问题1检索不到相关文档解决方案检查文本切分策略调整chunk_size和chunk_overlap参数优化查询改写实现查询扩展和同义词替换检查Embedding模型选择适合领域任务的模型# query_optimizer.py class QueryOptimizer: def __init__(self): self.synonyms { 价格: [价钱, 费用, 成本, 价位], 功能: [作用, 用途, 功效, 特性] } def expand_query(self, query: str) - List[str]: 查询扩展 expanded_queries [query] # 同义词扩展 for word, synonyms in self.synonyms.items(): if word in query: for synonym in synonyms: expanded_queries.append(query.replace(word, synonym)) return expanded_queries def spell_correct(self, query: str) - str: 拼写纠正 # 可以集成第三方拼写纠正库 return query问题2检索结果重复或碎片化解决方案实现结果去重基于语义相似度去重优化重排序使用更精细的reranker模型后处理聚合合并相关的文档片段7.2 性能瓶颈排查问题1响应时间过长排查步骤分析各组件耗时文档加载、向量化、检索、生成检查硬件资源CPU、内存、GPU使用率优化批量处理调整batch_size参数启用缓存对频繁查询结果进行缓存问题2内存占用过高优化策略流式处理大文档避免一次性加载所有文档使用内存映射文件减少内存占用定期清理缓存释放不必要的内存8. 生产环境最佳实践8.1 监控与日志完整的监控体系# monitoring.py import time import logging from datetime import datetime from prometheus_client import Counter, Histogram, start_http_server class Monitoring: def __init__(self): # 指标定义 self.query_counter Counter(rag_queries_total, Total queries) self.error_counter Counter(rag_errors_total, Total errors) self.response_time Histogram(rag_response_time, Response time) # 日志配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(rag_system.log), logging.StreamHandler() ] ) self.logger logging.getLogger(__name__) def log_query(self, question: str, response_time: float, success: bool True): 记录查询日志 self.query_counter.inc() self.response_time.observe(response_time) if not success: self.error_counter.inc() self.logger.info(fQuery: {question}, ResponseTime: {response_time:.2f}s, Success: {success}) # 使用示例 monitor Monitoring() def timed_query(question: str): start_time time.time() try: # 执行查询 result rag_system.query(question) response_time time.time() - start_time monitor.log_query(question, response_time, True) return result except Exception as e: response_time time.time() - start_time monitor.log_query(question, response_time, False) raise e8.2 安全与权限控制API安全防护# security.py from functools import wraps from flask import request, jsonify import jwt from datetime import datetime, timedelta class SecurityManager: def __init__(self, secret_key: str): self.secret_key secret_key def generate_token(self, user_id: str, expires_hours: int 24) - str: 生成JWT token payload { user_id: user_id, exp: datetime.utcnow()