开始技术细节之前,关于有声书想多絮叨两句(也算是一种产品需求分析):
基于上述的分析,打算用MCP Servers + LLM实现将电子书转换为有声书的小工具,使用方式很简单在AI交互框中输入“将xxx目录中xxx转为成音频”,这样我就拥有了一个低配版的喜马拉雅,将任意自己喜欢电子书变成自己和家人喜欢的有声书了。
具体实现比较简单,详细可以参考demo代码在GitHub仓库中 MCP agents,实现过程中我觉得比较有意思的两个细节可以分享一下:
我在进行demo测试的时候,用的是海明威的「老人与海」的英文版(约4万字),markitdown mcp调用之后返回大约6万字的markdown文本,正常情况下ReAct Agent会将mcp返回的内容作为LLM的上下文message进行下一轮的任务输入,此时就碰到问题了,明显超出了Qwen的上下文token限制(30,720),此时会直接导致Agent任务执行失败。
再回来分析markitdown mcp调用的结果,ReAct Agent并不需要大模型进行文本处理,所以我们只需要Agent直接返回mcp的结果即可。非常敬佩langchain(langraph)框架,这种细节需求场景都能覆盖到,虽然它饱受诟病。话不多说,直接上代码:
async def convert_epub_to_markdown(self, query: str) -> str:
messages = [{"role": "user", "content": query}]
available_tools = []
# Load tools from all sessions
for session in self.sessions:
tools = await load_mcp_tools(self.sessions[session])
available_tools.extend(tools)
if session == "markitdown-mcp":
# enforce a tool output from an agent avoiding any additional text the the agent adds after the tool call
for tool in tools:
tool.return_direct = True
# logger.debug("Available tools:", available_tools[0])
# Initial Qwen API call
agent = create_react_agent(self.qwen, available_tools)
response = await agent.ainvoke({"messages": query})
# Detailed processing of tool calls if you use other LLM, e.g. Claude
self.__md_str = response["messages"][-1].content
return response["messages"][-1].content
Cosyvoice LLM模型有一个限制,在进行文本转语音的时候发送的文本长度不得超过2000字符,这名相不满足「老人与海」有声书转换的要求,为此根据模型max length限制,我将超长文本按照语义和格式进行分割,然后批量进行语音转换。
def split_text(self, text: str, max_length: int = 2000) -> list:
splitter = MarkdownTextSplitter(chunk_size=max_length, chunk_overlap=0)
return splitter.split_text(text)
def exec():
# ...省略一些代码
await client.connect_to_mcp_server("markitdown-mcp")
# await client.connect_to_mcp_server("filesystem-mcp")
response = await client.convert_epub_to_markdown(
"把文件`/Users/edony/Downloads/hemingway-old-man-and-the-sea.epub`转换为markdown格式,大模型不要处理任何markdown,直接输出全部原文"
)
# split_text method
chunks = client.split_text(response, max_length=2000)
batch_size = 10
batch = math.ceil(len(chunks) / batch_size)
for i in range(batch):
sub_chunks = chunks[i * batch_size : (i + 1) * batch_size]
await client.synthesis_text_to_speech_using_asyncio(sub_chunks, i)
logger.info(len(chunks))