Nanbeige4.1-3B企业落地：内部知识库RAG增强+Agent工作流集成方案
·
Nanbeige4.1-3B企业落地：内部知识库RAG增强+Agent工作流集成方案
1. 引言：当小模型遇上大需求
想象一下这个场景：你是一家公司的技术负责人，每天要处理来自不同部门的几十个问题。销售部问你“我们产品的技术优势是什么？”，客服部需要“最新的客户投诉处理流程”，研发部则想知道“上个月项目评审会的关键结论”。你手头有堆积如山的文档、会议纪要、产品手册，但就是找不到最准确的答案。
这就是企业内部知识管理的典型痛点：信息分散、检索困难、响应缓慢。传统的搜索工具只能匹配关键词，无法理解问题的真正意图；而调用大模型API又面临成本高、数据安全、响应延迟等问题。
今天要介绍的 Nanbeige4.1-3B，就是为解决这类问题而生的。这个只有30亿参数的小模型，却拥有惊人的能力：支持8K长上下文、600步工具调用、强大的中文推理能力。更重要的是，它完全开源，可以部署在你自己的服务器上，数据不出域，安全可控。
本文将带你一步步实现一个完整的解决方案：用Nanbeige4.1-3B构建企业内部知识库问答系统，并结合Agent工作流实现自动化处理。无论你是技术工程师、产品经理还是企业决策者，都能从中找到可以直接落地的实用方案。
2. 为什么选择Nanbeige4.1-3B？
2.1 小身材，大能量
你可能会有疑问：现在动辄几百亿、上千亿参数的大模型那么多，为什么要选择一个只有30亿参数的小模型？
答案很简单：性价比和实用性。
让我们做个对比：
| 对比维度 | 百亿级大模型 | Nanbeige4.1-3B | 优势分析 |
|---|---|---|---|
| 部署成本 | 需要高端GPU，显存需求大 | 6GB显存即可运行，普通服务器就能部署 | 成本降低90%以上 |
| 响应速度 | 通常较慢，需要等待 | 本地推理，毫秒级响应 | 用户体验大幅提升 |
| 数据安全 | 数据需上传到云端 | 完全本地部署，数据不出域 | 满足企业安全合规要求 |
| 定制能力 | 通常只能调用API | 可以微调、定制、深度集成 | 灵活适应企业特定需求 |
| 工具调用 | 支持但成本高 | 支持600步长工具调用，业界领先 | 适合复杂工作流 |
2.2 核心技术特性解析
Nanbeige4.1-3B的几个核心特性，让它特别适合企业级应用：
8K长上下文：这意味着它可以一次性处理很长的文档。比如一份50页的产品说明书，你可以直接喂给它，它能够理解全文内容并准确回答问题。
600步工具调用：这是它的“杀手锏”。传统的工具调用可能只支持几十步，而600步意味着它可以执行非常复杂的多步骤任务。比如“从数据库查询数据→分析趋势→生成报告→发送邮件”这样的完整工作流。
强大的中文推理：基于23T高质量中文数据训练，在中文理解和生成方面表现出色。这对于中文环境的企业来说至关重要。
完全开源：模型权重、技术报告、训练数据全部开源，你可以完全掌控，不用担心供应商锁定问题。
3. 环境准备与快速部署
3.1 硬件与软件要求
在开始之前，我们先确认一下环境要求。别担心，要求并不高：
硬件要求：
- GPU：NVIDIA显卡，显存6GB以上（RTX 3060/4060等消费级显卡即可）
- 内存：16GB以上
- 存储：至少20GB可用空间
软件要求：
- 操作系统：Ubuntu 20.04/22.04或CentOS 7/8
- Python：3.8或更高版本
- CUDA：11.8或更高版本（如果使用GPU）
3.2 一键部署脚本
为了简化部署过程，我准备了一个完整的部署脚本。你只需要复制粘贴，就能完成所有环境配置：
#!/bin/bash
# nanbeige-deploy.sh - Nanbeige4.1-3B一键部署脚本
echo "开始部署Nanbeige4.1-3B企业知识库系统..."
# 1. 创建项目目录
mkdir -p /opt/nanbeige-enterprise
cd /opt/nanbeige-enterprise
# 2. 创建Python虚拟环境
echo "创建Python虚拟环境..."
python3 -m venv venv
source venv/bin/activate
# 3. 安装基础依赖
echo "安装基础依赖..."
pip install --upgrade pip
pip install torch==2.1.0 torchvision==0.16.0 torchaudio==2.1.0 --index-url https://download.pytorch.org/whl/cu118
pip install transformers==4.51.0 accelerate==0.20.0
# 4. 安装RAG相关依赖
echo "安装RAG相关依赖..."
pip install langchain==0.1.0 langchain-community==0.0.10
pip install chromadb==0.4.22 sentence-transformers==2.2.2
pip install pypdf==3.17.4 python-docx==1.1.0 markdown==3.5.2
# 5. 安装Web界面依赖
echo "安装Web界面依赖..."
pip install gradio==4.19.2 streamlit==1.28.0
# 6. 下载模型（如果已有模型文件可跳过）
echo "下载模型文件..."
# 这里需要从Hugging Face或官方渠道下载模型
# 假设模型已下载到 /root/ai-models/nanbeige/Nanbeige4___1-3B
echo "部署完成！"
echo "模型路径: /root/ai-models/nanbeige/Nanbeige4___1-3B"
echo "项目路径: /opt/nanbeige-enterprise"
3.3 验证安装
部署完成后，运行一个简单的测试脚本，确保一切正常：
# test_nanbeige.py - 验证模型加载和基础功能
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import time
def test_basic_inference():
"""测试基础推理功能"""
print("开始测试Nanbeige4.1-3B基础推理...")
# 模型路径
model_path = "/root/ai-models/nanbeige/Nanbeige4___1-3B"
# 记录开始时间
start_time = time.time()
# 加载模型和分词器
print("加载模型中...")
tokenizer = AutoTokenizer.from_pretrained(
model_path,
trust_remote_code=True
)
model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True
)
load_time = time.time() - start_time
print(f"模型加载完成，耗时: {load_time:.2f}秒")
# 测试对话
messages = [
{"role": "user", "content": "你好，请用一句话介绍你自己"}
]
input_ids = tokenizer.apply_chat_template(
messages,
return_tensors="pt"
).to(model.device)
# 生成回复
print("生成回复中...")
gen_start = time.time()
outputs = model.generate(
input_ids,
max_new_tokens=100,
temperature=0.7,
top_p=0.9,
do_sample=True
)
response = tokenizer.decode(
outputs[0][len(input_ids[0]):],
skip_special_tokens=True
)
gen_time = time.time() - gen_start
total_time = time.time() - start_time
print(f"回复: {response}")
print(f"生成耗时: {gen_time:.2f}秒")
print(f"总耗时: {total_time:.2f}秒")
return True
if __name__ == "__main__":
test_basic_inference()
运行这个脚本，如果看到类似下面的输出，说明部署成功：
开始测试Nanbeige4.1-3B基础推理...
加载模型中...
模型加载完成，耗时: 15.32秒
生成回复中...
回复: 我是Nanbeige4.1-3B，一个专注于中文理解和推理的开源语言模型。
生成耗时: 1.45秒
总耗时: 16.77秒
4. 构建企业知识库RAG系统
4.1 RAG系统架构设计
RAG（检索增强生成）的核心思想很简单：当用户提问时，先从知识库中找到相关的文档片段，然后把问题和这些片段一起交给模型，让模型基于这些信息生成答案。
我们的系统架构如下：
企业知识库RAG系统架构
├── 文档处理层
│ ├── 文档加载（PDF、Word、Excel、TXT等）
│ ├── 文本分割（按段落、按章节）
│ └── 向量化处理
├── 向量数据库层
│ ├── ChromaDB（轻量级向量数据库）
│ ├── 文档索引
│ └── 相似度检索
├── 模型推理层
│ ├── Nanbeige4.1-3B模型
│ ├── 提示词工程
│ └── 答案生成
└── 应用接口层
├── Web界面
├── API接口
└── 工作流集成
4.2 文档处理与向量化
首先，我们需要处理企业的各种文档。这里提供一个完整的文档处理模块：
# document_processor.py - 文档处理与向量化
import os
from typing import List, Dict, Any
from langchain.document_loaders import (
PyPDFLoader,
Docx2txtLoader,
TextLoader,
UnstructuredExcelLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
import chromadb
class EnterpriseDocumentProcessor:
"""企业文档处理器"""
def __init__(self, model_path: str = "/root/ai-models/nanbeige/Nanbeige4___1-3B"):
"""
初始化文档处理器
Args:
model_path: 模型路径，用于获取嵌入模型
"""
self.model_path = model_path
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每个片段1000字符
chunk_overlap=200, # 片段重叠200字符
length_function=len,
separators=["\n\n", "\n", "。", "；", "，", " ", ""]
)
# 使用中文优化的嵌入模型
self.embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5",
model_kwargs={'device': 'cuda'},
encode_kwargs={'normalize_embeddings': True}
)
# 向量数据库持久化路径
self.persist_directory = "./chroma_db"
def load_documents(self, file_paths: List[str]) -> List[Dict[str, Any]]:
"""
加载多种格式的文档
Args:
file_paths: 文件路径列表
Returns:
文档内容列表
"""
documents = []
for file_path in file_paths:
if not os.path.exists(file_path):
print(f"文件不存在: {file_path}")
continue
file_ext = os.path.splitext(file_path)[1].lower()
try:
if file_ext == '.pdf':
loader = PyPDFLoader(file_path)
elif file_ext == '.docx':
loader = Docx2txtLoader(file_path)
elif file_ext == '.txt':
loader = TextLoader(file_path, encoding='utf-8')
elif file_ext in ['.xlsx', '.xls']:
loader = UnstructuredExcelLoader(file_path)
else:
print(f"不支持的文件格式: {file_ext}")
continue
loaded_docs = loader.load()
for doc in loaded_docs:
documents.append({
'content': doc.page_content,
'metadata': {
'source': file_path,
'type': file_ext[1:], # 去掉点号
'page': doc.metadata.get('page', 0)
}
})
print(f"成功加载: {file_path}, 共{len(loaded_docs)}页")
except Exception as e:
print(f"加载文件失败 {file_path}: {str(e)}")
return documents
def split_documents(self, documents: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
分割文档为适合处理的片段
Args:
documents: 原始文档列表
Returns:
分割后的文档片段列表
"""
chunks = []
for doc in documents:
content = doc['content']
metadata = doc['metadata']
# 分割文本
text_chunks = self.text_splitter.split_text(content)
for i, chunk in enumerate(text_chunks):
chunk_metadata = metadata.copy()
chunk_metadata['chunk_id'] = i
chunk_metadata['total_chunks'] = len(text_chunks)
chunks.append({
'content': chunk,
'metadata': chunk_metadata
})
print(f"文档分割完成，共生成{len(chunks)}个片段")
return chunks
def create_vector_store(self, chunks: List[Dict[str, Any]], collection_name: str = "enterprise_knowledge"):
"""
创建向量存储
Args:
chunks: 文档片段列表
collection_name: 集合名称
Returns:
向量存储对象
"""
# 提取内容和元数据
texts = [chunk['content'] for chunk in chunks]
metadatas = [chunk['metadata'] for chunk in chunks]
# 创建向量数据库
vector_store = Chroma.from_texts(
texts=texts,
embedding=self.embeddings,
metadatas=metadatas,
persist_directory=self.persist_directory,
collection_name=collection_name
)
# 持久化到磁盘
vector_store.persist()
print(f"向量存储创建完成，保存到: {self.persist_directory}")
print(f"集合名称: {collection_name}, 文档数量: {len(texts)}")
return vector_store
def load_vector_store(self, collection_name: str = "enterprise_knowledge"):
"""
加载已有的向量存储
Args:
collection_name: 集合名称
Returns:
向量存储对象
"""
vector_store = Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embeddings,
collection_name=collection_name
)
# 获取集合中的文档数量
collection = vector_store._collection
count = collection.count()
print(f"加载向量存储成功，文档数量: {count}")
return vector_store
# 使用示例
if __name__ == "__main__":
# 初始化处理器
processor = EnterpriseDocumentProcessor()
# 1. 加载文档
documents = processor.load_documents([
"/path/to/company_handbook.pdf",
"/path/to/product_spec.docx",
"/path/to/meeting_notes.txt"
])
# 2. 分割文档
chunks = processor.split_documents(documents)
# 3. 创建向量存储
vector_store = processor.create_vector_store(chunks)
# 4. 测试检索
query = "公司的请假流程是什么？"
results = vector_store.similarity_search(query, k=3)
print(f"\n查询: {query}")
print("检索结果:")
for i, doc in enumerate(results):
print(f"\n[{i+1}] {doc.metadata['source']} (Page {doc.metadata.get('page', 'N/A')})")
print(f"内容: {doc.page_content[:200]}...")
4.3 智能检索与问答系统
有了向量化的知识库，接下来构建问答系统：
# qa_system.py - 智能问答系统
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from typing import List, Dict, Any
from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
class EnterpriseQASystem:
"""企业知识问答系统"""
def __init__(self, model_path: str, vector_store_path: str = "./chroma_db"):
"""
初始化问答系统
Args:
model_path: Nanbeige模型路径
vector_store_path: 向量数据库路径
"""
self.model_path = model_path
# 加载模型和分词器
print("加载问答模型...")
self.tokenizer = AutoTokenizer.from_pretrained(
model_path,
trust_remote_code=True
)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True
)
# 加载向量数据库
print("加载向量数据库...")
self.embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5",
model_kwargs={'device': 'cuda'},
encode_kwargs={'normalize_embeddings': True}
)
self.vector_store = Chroma(
persist_directory=vector_store_path,
embedding_function=self.embeddings,
collection_name="enterprise_knowledge"
)
# 系统提示词模板
self.system_prompt = """你是一个专业的企业知识助手，基于提供的上下文信息回答问题。
回答要求：
1. 只基于提供的上下文信息回答，不要编造信息
2. 如果上下文没有相关信息，如实告知"根据现有资料，无法回答此问题"
3. 回答要准确、简洁、专业
4. 如果涉及流程、步骤，请分点说明
上下文信息：
{context}
用户问题：{question}
请基于以上信息回答："""
def retrieve_context(self, question: str, k: int = 5) -> str:
"""
检索相关上下文
Args:
question: 用户问题
k: 返回的文档数量
Returns:
检索到的上下文
"""
# 相似度检索
docs = self.vector_store.similarity_search(question, k=k)
# 合并上下文
context_parts = []
for i, doc in enumerate(docs):
source = doc.metadata.get('source', '未知来源')
page = doc.metadata.get('page', 'N/A')
context_parts.append(f"[文档{i+1}] 来源: {source}, 页码: {page}")
context_parts.append(f"内容: {doc.page_content}")
context_parts.append("") # 空行分隔
return "\n".join(context_parts)
def generate_answer(self, question: str, max_tokens: int = 1024) -> Dict[str, Any]:
"""
生成答案
Args:
question: 用户问题
max_tokens: 最大生成token数
Returns:
包含答案和元数据的字典
"""
# 1. 检索相关上下文
print(f"检索问题: {question}")
context = self.retrieve_context(question)
# 2. 构建提示词
prompt = self.system_prompt.format(
context=context,
question=question
)
# 3. 准备模型输入
messages = [
{"role": "user", "content": prompt}
]
input_ids = self.tokenizer.apply_chat_template(
messages,
return_tensors="pt"
).to(self.model.device)
# 4. 生成答案
with torch.no_grad():
outputs = self.model.generate(
input_ids,
max_new_tokens=max_tokens,
temperature=0.7,
top_p=0.9,
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id
)
# 5. 解码答案
answer = self.tokenizer.decode(
outputs[0][len(input_ids[0]):],
skip_special_tokens=True
)
# 6. 提取引用来源
docs = self.vector_store.similarity_search(question, k=3)
sources = []
for doc in docs:
source_info = {
'source': doc.metadata.get('source', '未知'),
'page': doc.metadata.get('page', 'N/A'),
'content_preview': doc.page_content[:100] + '...'
}
sources.append(source_info)
return {
'question': question,
'answer': answer.strip(),
'sources': sources,
'context_used': len(context) > 0
}
def batch_qa(self, questions: List[str]) -> List[Dict[str, Any]]:
"""
批量问答
Args:
questions: 问题列表
Returns:
答案列表
"""
results = []
for question in questions:
try:
result = self.generate_answer(question)
results.append(result)
except Exception as e:
results.append({
'question': question,
'answer': f"处理问题时出错: {str(e)}",
'sources': [],
'context_used': False
})
return results
# 使用示例
if __name__ == "__main__":
# 初始化问答系统
qa_system = EnterpriseQASystem(
model_path="/root/ai-models/nanbeige/Nanbeige4___1-3B",
vector_store_path="./chroma_db"
)
# 测试问答
test_questions = [
"公司的年假政策是什么？",
"如何申请项目经费？",
"新员工入职需要准备哪些材料？"
]
print("开始测试问答系统...")
results = qa_system.batch_qa(test_questions)
for i, result in enumerate(results):
print(f"\n{'='*60}")
print(f"问题 {i+1}: {result['question']}")
print(f"答案: {result['answer']}")
if result['sources']:
print("\n参考来源:")
for source in result['sources']:
print(f" - {source['source']} (页码: {source['page']})")
print(f" 预览: {source['content_preview']}")
5. Agent工作流集成
5.1 Agent系统设计
Nanbeige4.1-3B支持600步工具调用，这意味着我们可以构建复杂的多步骤工作流。下面是一个完整的Agent系统实现：
# enterprise_agent.py - 企业智能Agent系统
import json
import requests
from typing import Dict, List, Any, Callable
from datetime import datetime
import sqlite3
class EnterpriseAgent:
"""企业智能Agent"""
def __init__(self, model_path: str):
"""
初始化Agent
Args:
model_path: 模型路径
"""
self.model_path = model_path
self.tools = self._register_tools()
self.conversation_history = []
def _register_tools(self) -> Dict[str, Dict[str, Any]]:
"""
注册可用工具
Returns:
工具字典
"""
tools = {
# 数据库查询工具
'query_database': {
'function': self.query_database,
'description': '查询企业数据库，获取员工、项目、客户等信息',
'parameters': {
'query_type': {
'type': 'string',
'enum': ['employee', 'project', 'customer', 'finance'],
'description': '查询类型'
},
'filters': {
'type': 'object',
'description': '查询过滤条件'
},
'limit': {
'type': 'integer',
'description': '返回结果数量限制'
}
}
},
# 邮件发送工具
'send_email': {
'function': self.send_email,
'description': '发送电子邮件',
'parameters': {
'to': {
'type': 'array',
'items': {'type': 'string'},
'description': '收件人邮箱列表'
},
'subject': {
'type': 'string',
'description': '邮件主题'
},
'content': {
'type': 'string',
'description': '邮件内容'
},
'attachments': {
'type': 'array',
'items': {'type': 'string'},
'description': '附件路径列表'
}
}
},
# 日程管理工具
'manage_calendar': {
'function': self.manage_calendar,
'description': '管理日历事件',
'parameters': {
'action': {
'type': 'string',
'enum': ['create', 'update', 'delete', 'query'],
'description': '操作类型'
},
'event_id': {
'type': 'string',
'description': '事件ID（更新/删除时需提供）'
},
'title': {
'type': 'string',
'description': '事件标题'
},
'start_time': {
'type': 'string',
'description': '开始时间（格式: YYYY-MM-DD HH:MM）'
},
'end_time': {
'type': 'string',
'description': '结束时间（格式: YYYY-MM-DD HH:MM）'
},
'participants': {
'type': 'array',
'items': {'type': 'string'},
'description': '参与者邮箱列表'
}
}
},
# 文档生成工具
'generate_document': {
'function': self.generate_document,
'description': '生成各种文档',
'parameters': {
'doc_type': {
'type': 'string',
'enum': ['report', 'proposal', 'meeting_minutes', 'email'],
'description': '文档类型'
},
'topic': {
'type': 'string',
'description': '文档主题'
},
'content_points': {
'type': 'array',
'items': {'type': 'string'},
'description': '内容要点'
},
'format': {
'type': 'string',
'enum': ['markdown', 'html', 'plain_text'],
'description': '输出格式'
}
}
},
# 数据分析工具
'analyze_data': {
'function': self.analyze_data,
'description': '分析数据并生成洞察',
'parameters': {
'data_source': {
'type': 'string',
'description': '数据源（文件路径或数据库表名）'
},
'analysis_type': {
'type': 'string',
'enum': ['summary', 'trend', 'comparison', 'prediction'],
'description': '分析类型'
},
'metrics': {
'type': 'array',
'items': {'type': 'string'},
'description': '分析指标'
},
'time_range': {
'type': 'object',
'description': '时间范围'
}
}
}
}
return tools
def query_database(self, query_type: str, filters: Dict = None, limit: int = 10) -> Dict:
"""查询数据库工具"""
# 这里连接实际的企业数据库
# 示例实现使用SQLite模拟
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
# 创建示例表（实际应用中替换为真实表）
if query_type == 'employee':
cursor.execute('''
CREATE TABLE IF NOT EXISTS employees (
id INTEGER PRIMARY KEY,
name TEXT,
department TEXT,
position TEXT,
hire_date TEXT,
salary REAL
)
''')
# 插入示例数据
cursor.executemany('''
INSERT INTO employees (name, department, position, hire_date, salary)
VALUES (?, ?, ?, ?, ?)
''', [
('张三', '技术部', '高级工程师', '2022-03-15', 25000),
('李四', '市场部', '市场经理', '2021-08-20', 18000),
('王五', '人事部', 'HR专员', '2023-01-10', 12000),
('赵六', '财务部', '财务主管', '2020-11-05', 22000)
])
# 执行查询
query = "SELECT * FROM employees"
if filters:
conditions = []
for key, value in filters.items():
conditions.append(f"{key} = '{value}'")
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += f" LIMIT {limit}"
cursor.execute(query)
elif query_type == 'project':
# 类似实现其他查询类型
pass
results = cursor.fetchall()
conn.close()
return {
'status': 'success',
'query_type': query_type,
'results': results,
'count': len(results)
}
def send_email(self, to: List[str], subject: str, content: str, attachments: List[str] = None) -> Dict:
"""发送邮件工具"""
# 这里集成实际的邮件发送服务
# 示例实现只记录日志
email_data = {
'to': to,
'subject': subject,
'content': content,
'attachments': attachments or [],
'sent_time': datetime.now().isoformat()
}
# 实际发送邮件的代码
# import smtplib
# from email.mime.text import MIMEText
# from email.mime.multipart import MIMEMultipart
print(f"[邮件发送] 收件人: {to}")
print(f"[邮件发送] 主题: {subject}")
print(f"[邮件发送] 内容: {content[:100]}...")
return {
'status': 'success',
'message': '邮件发送成功',
'data': email_data
}
def manage_calendar(self, action: str, **kwargs) -> Dict:
"""日历管理工具"""
# 这里集成实际的日历API（如Google Calendar、Outlook等）
if action == 'create':
event_data = {
'event_id': f"event_{datetime.now().timestamp()}",
'title': kwargs.get('title', '未命名事件'),
'start_time': kwargs.get('start_time'),
'end_time': kwargs.get('end_time'),
'participants': kwargs.get('participants', []),
'created_at': datetime.now().isoformat()
}
print(f"[日历] 创建事件: {event_data['title']}")
return {
'status': 'success',
'action': 'create',
'event': event_data
}
# 其他操作类似实现
return {
'status': 'success',
'action': action,
'message': f'日历操作 {action} 完成'
}
def generate_document(self, doc_type: str, topic: str, content_points: List[str], format: str = 'markdown') -> Dict:
"""文档生成工具"""
# 使用模型生成文档内容
prompt = f"""请生成一份{doc_type}，主题是：{topic}
内容要点：
{chr(10).join(f'- {point}' for point in content_points)}
请按照{format}格式生成完整文档。"""
# 这里调用模型生成文档
# 实际实现中会调用Nanbeige模型
document_content = f"""# {topic}
## 概述
这是一份关于{topic}的{doc_type}。
## 主要内容
{chr(10).join(f'1. {point}' for point in content_points)}
## 总结
以上是{doc_type}的主要内容。"""
return {
'status': 'success',
'doc_type': doc_type,
'topic': topic,
'content': document_content,
'format': format,
'generated_at': datetime.now().isoformat()
}
def analyze_data(self, data_source: str, analysis_type: str, metrics: List[str], time_range: Dict = None) -> Dict:
"""数据分析工具"""
# 这里集成实际的数据分析逻辑
analysis_result = {
'data_source': data_source,
'analysis_type': analysis_type,
'metrics': metrics,
'time_range': time_range or {},
'insights': [
f"指标 {metric} 在指定时间范围内呈现增长趋势"
for metric in metrics
],
'recommendations': [
"建议继续关注核心指标变化",
"考虑优化相关业务流程"
],
'generated_at': datetime.now().isoformat()
}
return {
'status': 'success',
'analysis': analysis_result
}
def execute_workflow(self, user_request: str, max_steps: int = 10) -> Dict:
"""
执行工作流
Args:
user_request: 用户请求
max_steps: 最大执行步骤
Returns:
执行结果
"""
print(f"\n开始处理请求: {user_request}")
# 1. 分析用户请求，规划工作流
workflow_plan = self._plan_workflow(user_request)
print(f"工作流规划: {workflow_plan}")
# 2. 按步骤执行
execution_steps = []
results = []
for step in workflow_plan[:max_steps]:
print(f"\n执行步骤: {step}")
try:
tool_name = step.get('tool')
tool_config = self.tools.get(tool_name)
if not tool_config:
raise ValueError(f"工具不存在: {tool_name}")
# 执行工具
tool_function = tool_config['function']
tool_params = step.get('parameters', {})
result = tool_function(**tool_params)
execution_steps.append({
'step': step,
'status': 'success',
'result': result
})
results.append(result)
print(f"步骤完成，结果: {result.get('status', 'unknown')}")
except Exception as e:
execution_steps.append({
'step': step,
'status': 'error',
'error': str(e)
})
print(f"步骤失败: {str(e)}")
# 3. 汇总结果
final_result = self._summarize_results(user_request, execution_steps, results)
return {
'user_request': user_request,
'workflow_plan': workflow_plan,
'execution_steps': execution_steps,
'final_result': final_result,
'total_steps': len(execution_steps),
'successful_steps': len([s for s in execution_steps if s['status'] == 'success'])
}
def _plan_workflow(self, user_request: str) -> List[Dict]:
"""
规划工作流步骤
Args:
user_request: 用户请求
Returns:
工作流步骤列表
"""
# 这里可以集成更智能的规划逻辑
# 示例实现基于关键词匹配
workflow = []
# 示例：如果请求包含"报告"和"数据"，先查询数据再生成报告
if "报告" in user_request and "数据" in user_request:
workflow.append({
'tool': 'query_database',
'description': '查询相关数据',
'parameters': {
'query_type': 'project',
'filters': {},
'limit': 100
}
})
workflow.append({
'tool': 'analyze_data',
'description': '分析数据趋势',
'parameters': {
'data_source': 'project_data',
'analysis_type': 'trend',
'metrics': ['progress', 'budget', 'timeline']
}
})
workflow.append({
'tool': 'generate_document',
'description': '生成分析报告',
'parameters': {
'doc_type': 'report',
'topic': '项目数据分析报告',
'content_points': ['数据概览', '趋势分析', '建议措施'],
'format': 'markdown'
}
})
if "发送" in user_request or "邮件" in user_request:
workflow.append({
'tool': 'send_email',
'description': '发送报告邮件',
'parameters': {
'to': ['manager@company.com'],
'subject': '项目数据分析报告',
'content': '附件是项目数据分析报告，请查收。'
}
})
# 其他场景的工作流规划...
return workflow
def _summarize_results(self, user_request: str, steps: List[Dict], results: List[Any]) -> Dict:
"""
汇总工作流结果
Args:
user_request: 原始请求
steps: 执行步骤
results: 各步骤结果
Returns:
汇总结果
"""
successful_steps = [s for s in steps if s['status'] == 'success']
summary = {
'original_request': user_request,
'total_steps': len(steps),
'successful_steps': len(successful_steps),
'failed_steps': len(steps) - len(successful_steps),
'key_results': [],
'completion_status': '部分完成' if len(successful_steps) < len(steps) else '全部完成'
}
# 提取关键结果
for step, result in zip(steps, results):
if step['status'] == 'success' and result:
tool_name = step['step'].get('tool', 'unknown')
summary['key_results'].append({
'tool': tool_name,
'result_summary': str(result)[:200] # 截取前200字符
})
return summary
# 使用示例
if __name__ == "__main__":
# 初始化Agent
agent = EnterpriseAgent(
model_path="/root/ai-models/nanbeige/Nanbeige4___1-3B"
)
# 测试工作流执行
test_requests = [
"帮我查询技术部的员工信息，然后生成一份部门报告，最后发送给人事经理",
"分析上个季度的项目数据，生成趋势分析报告",
"为下周的团队会议创建日历事件，并发送会议邀请"
]
for request in test_requests:
print(f"\n{'='*60}")
print(f"处理请求: {request}")
result = agent.execute_workflow(request, max_steps=5)
print(f"\n处理完成!")
print(f"总步骤: {result['total_steps']}")
print(f"成功步骤: {result['successful_steps']}")
print(f"完成状态: {result['final_result']['completion_status']}")
if result['final_result']['key_results']:
print("\n关键结果:")
for kr in result['final_result']['key_results']:
print(f" - {kr['tool']}: {kr['result_summary']}")
5.2 工作流编排与自动化
基于上面的Agent系统，我们可以构建更复杂的工作流编排：
# workflow_orchestrator.py - 工作流编排器
import yaml
import json
from typing import Dict, List, Any
from datetime import datetime
from enum import Enum
class WorkflowStatus(Enum):
"""工作流状态枚举"""
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
PAUSED = "paused"
class WorkflowOrchestrator:
"""工作流编排器"""
def __init__(self, agent_system):
"""
初始化编排器
Args:
agent_system: Agent系统实例
"""
self.agent = agent_system
self.workflows = {}
self.execution_history = []
def load_workflow_template(self, template_path: str) -> Dict:
"""
加载工作流模板
Args:
template_path: 模板文件路径
Returns:
工作流配置
"""
with open(template_path, 'r', encoding='utf-8') as f:
template = yaml.safe_load(f)
workflow_id = f"workflow_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
workflow_config = {
'id': workflow_id,
'name': template.get('name', '未命名工作流'),
'description': template.get('description', ''),
'version': template.get('version', '1.0'),
'created_at': datetime.now().isoformat(),
'steps': template.get('steps', []),
'triggers': template.get('triggers', []),
'variables': template.get('variables', {}),
'status': WorkflowStatus.PENDING.value,
'current_step': 0,
'results': {}
}
self.workflows[workflow_id] = workflow_config
return workflow_config
def execute_workflow(self, workflow_id: str, input_data: Dict = None) -> Dict:
"""
执行工作流
Args:
workflow_id: 工作流ID
input_data: 输入数据
Returns:
执行结果
"""
if workflow_id not in self.workflows:
raise ValueError(f"工作流不存在: {workflow_id}")
workflow = self.workflows[workflow_id]
workflow['status'] = WorkflowStatus.RUNNING.value
workflow['started_at'] = datetime.now().isoformat()
print(f"开始执行工作流: {workflow['name']}")
execution_result = {
'workflow_id': workflow_id,
'workflow_name': workflow['name'],
'start_time': workflow['started_at'],
'steps_executed': [],
'final_output': None,
'errors': []
}
# 处理输入变量
variables = workflow['variables'].copy()
if input_data:
variables.update(input_data)
# 按顺序执行步骤
for step_index, step_config in enumerate(workflow['steps']):
try:
print(f"\n执行步骤 {step_index + 1}: {step_config.get('name', '未命名步骤')}")
# 解析步骤配置
step_type = step_config.get('type', 'agent_action')
step_name = step_config.get('name', f'step_{step_index}')
step_description = step_config.get('description', '')
# 替换变量
step_config = self._replace_variables(step_config, variables)
# 执行步骤
if step_type == 'agent_action':
step_result = self._execute_agent_step(step_config, variables)
elif step_type == 'condition':
step_result = self._execute_condition_step(step_config, variables)
elif step_type == 'loop':
step_result = self._execute_loop_step(step_config, variables)
elif step_type == 'data_transform':
step_result = self._execute_data_transform_step(step_config, variables)
else:
raise ValueError(f"不支持的步骤类型: {step_type}")
# 保存步骤结果
step_execution = {
'step_index': step_index,
'step_name': step_name,
'step_type': step_type,
'status': 'success',
'result': step_result,
'executed_at': datetime.now().isoformat()
}
execution_result['steps_executed'].append(step_execution)
# 更新变量
if 'output_variable' in step_config:
var_name = step_config['output_variable']
variables[var_name] = step_result
print(f"设置变量 {var_name} = {step_result}")
workflow['current_step'] = step_index + 1
except Exception as e:
error_msg = f"步骤 {step_index + 1} 执行失败: {str(e)}"
print(error_msg)
step_execution = {
'step_index': step_index,
'step_name': step_config.get('name', f'step_{step_index}'),
'step_type': step_config.get('type', 'unknown'),
'status': 'failed',
'error': str(e),
'executed_at': datetime.now().isoformat()
}
execution_result['steps_executed'].append(step_execution)
execution_result['errors'].append(error_msg)
# 根据错误处理策略决定是否继续
error_handling = step_config.get('error_handling', 'stop')
if error_handling == 'stop':
workflow['status'] = WorkflowStatus.FAILED.value
break
elif error_handling == 'continue':
continue
elif error_handling == 'retry':
# 重试逻辑
max_retries = step_config.get('max_retries', 3)
retry_count = 0
while retry_count < max_retries:
try:
retry_count += 1
print(f"重试步骤 {step_index + 1} (第{retry_count}次)")
# 重新执行步骤
# ... 重试逻辑
break
except Exception as retry_error:
if retry_count == max_retries:
workflow['status'] = WorkflowStatus.FAILED.value
break
# 完成工作流
if workflow['status'] != WorkflowStatus.FAILED.value:
workflow['status'] = WorkflowStatus.COMPLETED.value
workflow['completed_at'] = datetime.now().isoformat()
# 生成最终输出
if 'output' in workflow:
execution_result['final_output'] = self._replace_variables(
workflow['output'], variables
)
workflow['results'] = execution_result
self.execution_history.append(execution_result)
print(f"\n工作流执行完成: {workflow['name']}")
print(f"状态: {workflow['status']}")
print(f"执行步骤: {len(execution_result['steps_executed'])}")
print(f"错误数: {len(execution_result['errors'])}")
return execution_result
def _execute_agent_step(self, step_config: Dict, variables: Dict) -> Any:
"""执行Agent步骤"""
tool_name = step_config.get('tool')
tool_params = step_config.get('parameters', {})
# 获取工具函数
tool_config = self.agent.tools.get(tool_name)
if not tool_config:
raise ValueError(f"Agent工具不存在: {tool_name}")
# 执行工具
tool_function = tool_config['function']
result = tool_function(**tool_params)
return result
def _execute_condition_step(self, step_config: Dict, variables: Dict) -> Any:
"""执行条件步骤"""
condition = step_config.get('condition', '')
true_branch = step_config.get('true_branch', [])
false_branch = step_config.get('false_branch', [])
# 评估条件（这里简化处理，实际需要更复杂的表达式求值）
condition_result = self._evaluate_condition(condition, variables)
if condition_result:
print(f"条件成立，执行真分支 ({len(true_branch)}个步骤)")
# 执行真分支步骤
for branch_step in true_branch:
self._execute_branch_step(branch_step, variables)
else:
print(f"条件不成立，执行假分支 ({len(false_branch)}个步骤)")
# 执行假分支步骤
for branch_step in false_branch:
self._execute_branch_step(branch_step, variables)
return condition_result
def _execute_loop_step(self, step_config: Dict, variables: Dict) -> List:
"""执行循环步骤"""
loop_var = step_config.get('loop_variable', 'item')
collection = step_config.get('collection', [])
steps = step_config.get('steps', [])
results = []
for item in collection:
# 设置循环变量
variables[loop_var] = item
print(f"循环处理: {loop_var} = {item}")
# 执行循环体内的步骤
for step in steps:
step_result = self._execute_branch_step(step, variables)
results.append(step_result)
return results
def _execute_data_transform_step(self, step_config: Dict, variables: Dict) -> Any:
"""执行数据转换步骤"""
transform_type = step_config.get('transform_type')
input_data = step_config.get('input')
if transform_type == 'filter':
condition = step_config.get('condition')
filtered = [item for item in input_data if self._evaluate_condition(condition, {'item': item})]
return filtered
elif transform_type == 'map':
expression = step_config.get('expression')
mapped = [eval(expression, {'item': item}) for item in input_data]
return mapped
elif transform_type == 'aggregate':
operation = step_config.get('operation')
if operation == 'sum':
return sum(input_data)
elif operation == 'average':
return sum(input_data) / len(input_data) if input_data else 0
elif operation == 'count':
return len(input_data)
elif operation == 'max':
return max(input_data) if input_data else None
elif operation == 'min':
return min(input_data) if input_data else None
return input_data
def _execute_branch_step(self, step_config: Dict, variables: Dict) -> Any:
"""执行分支步骤（内部使用）"""
step_type = step_config.get('type', 'agent_action')
if step_type == 'agent_action':
return self._execute_agent_step(step_config, variables)
elif step_type == 'data_transform':
return self._execute_data_transform_step(step_config, variables)
else:
# 其他类型的步骤...
return None
def _replace_variables(self, config: Any, variables: Dict) -> Any:
"""替换配置中的变量"""
if isinstance(config, str):
# 替换字符串中的变量占位符 ${var}
import re
pattern = r'\$\{(\w+)\}'
def replace_match(match):
var_name = match.group(1)
return str(variables.get(var_name, match.group(0)))
return re.sub(pattern, replace_match, config)
elif isinstance(config, dict):
# 递归处理字典
return {k: self._replace_variables(v, variables) for k, v in config.items()}
elif isinstance(config, list):
# 递归处理列表
return [self._replace_variables(item, variables) for item in config]
else:
return config
def _evaluate_condition(self, condition: str, variables: Dict) -> bool:
"""评估条件表达式"""
# 这里简化实现，实际需要更安全的表达式求值
try:
# 将变量注入到求值环境
eval_env = variables.copy()
# 添加一些常用的函数
eval_env.update({
'len': len,
'str': str,
'int': int,
'float': float,
'bool': bool
})
# 安全求值（实际生产环境应该使用更安全的方法）
result = eval(condition, {"__builtins__": {}}, eval_env)
return bool(result)
except Exception as e:
print(f"条件求值失败: {condition}, 错误: {str(e)}")
return False
def get_workflow_status(self, workflow_id: str) -> Dict:
"""获取工作流状态"""
if workflow_id not in self.workflows:
raise ValueError(f"工作流不存在: {workflow_id}")
workflow = self.workflows[workflow_id]
return {
'workflow_id': workflow_id,
'name': workflow['name'],
'status': workflow['status'],
'current_step': workflow['current_step'],
'total_steps': len(workflow['steps']),
'created_at': workflow['created_at'],
'started_at': workflow.get('started_at'),
'completed_at': workflow.get('completed_at')
}
def list_workflows(self) -> List[Dict]:
"""列出所有工作流"""
workflows_list = []
for workflow_id, workflow in self.workflows.items():
workflows_list.append({
'id': workflow_id,
'name': workflow['name'],
'description': workflow['description'],
'status': workflow['status'],
'current_step': workflow['current_step'],
'total_steps': len(workflow['steps']),
'created_at': workflow['created_at']
})
return workflows_list
# 工作流模板示例 (workflow_template.yaml)
workflow_template_example = """
name: "月度报告生成工作流"
description: "自动生成并发送月度部门报告"
version: "1.0"
variables:
department: "技术部"
report_month: "2024-01"
recipient_email: "manager@company.com"
triggers:
- type: "schedule"
cron: "0 0 1 * *" # 每月1日0点执行
- type: "manual"
description: "手动触发"
steps:
- name: "查询部门数据"
type: "agent_action"
tool: "query_database"
parameters:
query_type: "employee"
filters:
department: "${department}"
limit: 100
output_variable: "department_data"
error_handling: "retry"
max_retries: 3
- name: "分析员工数据"
type: "data_transform"
transform_type: "aggregate"
input: "${department_data.results}"
operation: "count"
output_variable: "employee_count"
- name: "生成报告"
type: "agent_action"
tool: "generate_document"
parameters:
doc_type: "report"
topic: "${department} ${report_month}月度报告"
content_points:
- "部门: ${department}"
- "报告月份: ${report_month}"
- "员工总数: ${employee_count}"
- "本月重点工作总结"
- "下月工作计划"
format: "markdown"
output_variable: "report_content"
- name: "发送报告"
type: "agent_action"
tool: "send_email"
parameters:
to:
- "${recipient_email}"
subject: "${department} ${report_month}月度报告"
content: |
您好，
附件是${department}${report_month}的月度工作报告，请查收。
如有任何问题，请随时联系。
此致
敬礼
attachments:
- "/reports/${department}_${report_month}.md"
error_handling: "continue"
- name: "记录执行日志"
type: "agent_action"
tool: "query_database"
parameters:
query_type: "system_log"
filters:
workflow_name: "${name}"
execution_time: "${started_at}"
limit: 1
output:
message: "月度报告生成完成"
report_generated: true
email_sent: true
execution_time: "${completed_at}"
"""
# 使用示例
if __name__ == "__main__":
# 1. 初始化Agent系统
agent = EnterpriseAgent(
model_path="/root/ai-models/nanbeige/Nanbeige4___1-3B"
)
# 2. 初始化工作流编排器
orchestrator = WorkflowOrchestrator(agent)
# 3. 加载工作流模板（实际从文件加载）
import tempfile
import os
# 创建临时模板文件
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
f.write(workflow_template_example)
template_path = f.name
try:
# 加载工作流
workflow_config = orchestrator.load_workflow_template(template_path)
print(f"加载工作流: {workflow_config['name']}")
print(f"工作流ID: {workflow_config['id']}")
# 4. 执行工作流
input_data = {
'department': '技术部',
'report_month': '2024-01',
'recipient_email': 'tech_manager@company.com'
}
result = orchestrator.execute_workflow(workflow_config['id'], input_data)
# 5. 查看结果
print(f"\n执行结果:")
print(f"工作流状态: {result.get('final_output', {}).get('message', 'N/A')}")
print(f"执行步骤数: {len(result['steps_executed'])}")
if result['errors']:
print(f"错误信息:")
for error in result['errors']:
print(f" - {error}")
# 6. 查看所有工作流状态
print(f"\n所有工作流状态:")
workflows = orchestrator.list_workflows()
for wf in workflows:
print(f" - {wf['name']}: {wf['status']} (进度: {wf['current_step']}/{wf['total_steps']})")
finally:
# 清理临时文件
os.unlink(template_path)
6. 系统集成与部署方案
6.1 完整的系统架构
现在我们已经有了各个组件，让我们把它们整合成一个完整的系统：
# enterprise_ai_system.py - 完整的企业AI系统
import uvicorn
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import logging
from datetime import datetime
import asyncio
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# 数据模型
class QueryRequest(BaseModel):
"""查询请求"""
question: str
user_id: Optional[str] = None
session_id: Optional[str] = None
context: Optional[Dict] = None
class QueryResponse(BaseModel):
"""查询响应"""
answer: str
sources: List[Dict[str, Any]]
processing_time: float
model_used: str
timestamp: str
class WorkflowRequest(BaseModel):
"""工作流请求"""
workflow_template: Dict[str, Any]
input_data: Dict[str, Any]
priority: Optional[str] = "normal"
class WorkflowResponse(BaseModel):
"""工作流响应"""
workflow_id: str
status: str
message: str
execution_id: Optional[str] = None
estimated_completion: Optional[str] = None
class EnterpriseAISystem:
"""企业AI系统（整合RAG + Agent）"""
def __init__(self, config_path: str = "config.yaml"):
"""
初始化企业AI系统
Args:
config_path: 配置文件路径
"""
self.config = self._load_config(config_path)
self.qa_system = None
self.agent_system = None
self.workflow_orchestrator = None
self._init_components()
# 会话管理
self.sessions = {}
# 异步任务队列
self.task_queue = asyncio.Queue()
self.worker_task = None
logger.info("企业AI系统初始化完成")
def _load_config(self, config_path: str) -> Dict:
"""加载配置文件"""
import yaml
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
logger.info(f"配置文件加载成功: {config_path}")
return config
except FileNotFoundError:
logger.warning(f"配置文件不存在: {config_path}，使用默认配置")
return self._get_default_config()
def _get_default_config(self) -> Dict:
"""获取默认配置"""
return {
'model': {
'path': '/root/ai-models/nanbeige/Nanbeige4___1-3B',
'device': 'auto',
'dtype': 'bfloat16'
},
'rag': {
'vector_store_path': './chroma_db',
'collection_name': 'enterprise_knowledge',
'retrieval_top_k': 5,
'similarity_threshold': 0.7
},
'agent': {
'max_steps': 20,
'timeout': 300 # 5分钟
},
'api': {
'host': '0.0.0.0',
'port': 8000,
'workers': 4,
'log_level': 'info'
},
'storage': {
'session_ttl': 3600, # 1小时
'max_sessions': 1000
}
}
def _init_components(self):
"""初始化各个组件"""
logger.info("初始化RAG问答系统...")
from qa_system import EnterpriseQASystem
self.qa_system = EnterpriseQASystem(
model_path=self.config['model']['path'],
vector_store_path=self.config['rag']['vector_store_path']
)
logger.info("初始化Agent系统...")
from enterprise_agent import EnterpriseAgent
self.agent_system = EnterpriseAgent(
model_path=self.config['model']['path']
)
logger.info("初始化工作流编排器...")
from workflow_orchestrator import WorkflowOrchestrator
self.workflow_orchestrator = WorkflowOrchestrator(self.agent_system)
logger.info("所有组件初始化完成")
async def start_async_worker(self):
"""启动异步工作线程"""
logger.info("启动异步工作线程...")
self.worker_task = asyncio.create_task(self._process_task_queue())
async def _process_task_queue(self):
"""处理任务队列"""
while True:
try:
task = await self.task_queue.get()
await self._execute_task(task)
self.task_queue.task_done()
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"任务处理失败: {str(e)}")
async def _execute_task(self, task: Dict):
"""执行任务"""
task_type = task.get('type')
task_id = task.get('id')
logger.info(f"开始执行任务: {task_id} ({task_type})")
try:
if task_type == 'workflow':
workflow_id = task.get('workflow_id')
input_data = task.get('input_data', {})
# 执行工作流
result = self.workflow_orchestrator.execute_workflow(
更多推荐
3
0- 0
已为社区贡献27条内容
所有评论(0)