AI智能体开发技术 教案全套 01-AI智能体开发技术-15-AI智能问答系统（二）

AI智能体开发技术一、什么是AI智能体AI智能体是一种具备自主性、目标导向的应用，它赋予了AI模型一种类似人类使用工具的能力。通过将模型的推理、逻辑判断与外部工具相连接，智能体能够自主规划、执行并调整其行动，以实现既定目标。一个典型的Agent由以下几个关键部分组成：1、大脑(Brain)-LLM：这是Agent的核心引擎，负责理解、推理、规划和决策。Agent的所有“思考”过程都由LLM完成。2、规划(Planning)：面对一个复杂任务，Agent需要将其分解为一系列可执行的小步骤。（1）任务分解(TaskDecomposition)：将宏大目标（如“组织一次团队旅行”）分解为具体子任务（“查询机票”、“预订机票”、“预订酒店”、“规划行程”等）。（2）自我反思(Self-Reflection)：Agent会评估自己上一步行动的结果，并根据结果修正下一步的计划。3、工具(Tools)：这是Agent与外界交互的“手和脚”。工具本质上就是Agent可以调用的函数或API（FunctionCalling、MCP、A2A等）。（1）示例：搜索引擎API、计算器、数据库查询接口、发送邮件的函数、天气查询API等。（2）工具选择：LLM根据当前任务和上下文，从其“工具箱”中选择最合适的工具来执行。4、记忆(Memory)：为了执行长期和复杂的任务，Agent需要记忆。（1）短期记忆：通常指当前的对话历史和上下文窗口，帮助Agent保持对话连贯。长期记忆：Agent可以将关键信息、过去的经验和学习成果存储在外部数据库（如向量数据库）中，以便在未来执行类似任务时参考，实现“经验积累”。二、智能体案例1、AI买咖啡2、AI生成PPTttps:///利用讯飞智文收费版本，基于原始文本生成PPT，效果相当不错。3、文生图（1）、先在DeepSeek中生成提示词1 1 一只中国的功夫熊猫在天上与一架F35战斗机决战，场景恢弘而激烈，天上乌云密布，大有一战到底的决心。请基于上述场景为我生成文生图的提示词。ttps:///（2）、在即梦中基于DeepSeek的提示词生成图片ttps:///ai-tool/home/4、海螺短视频生成（1）、先在DeepSeek中描述场景并生成提示词：图片中是一个视频生成提示词的公式，请你学习这个公式后面帮我写提示词的时候要按照这个公式来写，公式中每个部分都可以非常详细但总体要符合这个公式。图片中是一个视频生成提示词的公式，请你学习这个公式后面帮我写提示词的时候要按照这个公式来写，公式中每个部分都可以非常详细但总体要符合这个公式。现在请你仔细赏鉴李清照《如梦令》这首诗，提取这首诗想要表达的意境和每一句诗(一个标点一句)的灵魂，然后为每一句诗(一个标点一句)设计一个视频分镜，写出每个分镜的视频生成提示词，要求是:1、每个分镜提示词中都包含同一个女主角，这个女主角是中国古风少女形象，有着鲜明的形象特征，并贯穿每句诗。2、每个提示词最后加“中国古代卡通，明亮的色彩，3d皮克斯风格”。3、提示词要通俗易读，不要包含复杂的描述，不要有形容词、抒情、数字等。4、根据诗句含义确定每个分镜除了女主角，还应该包含哪些事物、详细场景画面、人物的行为、事物的动态等，根据上述信息确定合适的镜头运动，最终汇总到提示词中输出。5、提示词包含需要的所有信息，符合提示词公式，汇总成一段输出，中文。最终结果以表格形式输出，第一列是诗句，第二列是完整详细的提示词。（2）、再在海螺视频中生成视频ttps:///create5、在即梦中生成数字人需要先录制一段视频作为模板，录制一段音频作为音色：6、蜗牛学苑AI智慧课堂每日讲师上传视频后，将当天的视频生成语音，从语音中识别文字，再将文字内容进行摘要，生成思维导图，并基于当天的内容进行出题，考试后进行智能阅卷，对学习情况进行评估等，整个过程全流程自动化。7、蜗牛学苑AI面试官蜗牛学苑历年的面试数据加持，真人通过语音与AI面试官交流，AI提问，真人回答，最后生成面试评估报告：8、在线智能客服基于知识库，将蜗牛学苑的文档导入给系统，系统可以回答用户提的有关产品与服务的问题。9、AI质量检测10、半导体行业良率提升11、AI编程助手编程助手目前主流的有：Trae、Cline、Continue、Copilot、Cursor、Devin、通义灵码、Codex等，支持各类IDE，模型主要以GPT、DeepSeek、QWen、Gemini和Claude为主。一般编程助手可以实现的功能有：需求理解、文生代码、图生代码、代码理解、代码优化、代码审计、生成测试代码、代码搜索等。12、产品原型图生成直接根据需求文档生成产品原型图，非常适用于售前和需求讨论等工作，效率大幅提升。https://v0.de/13、OpenClaw智能体平台类似的平台还有KimiClaw、MaxClaw、CozeClaw、NuwaClaw等，或者类似的低代码平台Coze、Dify等。三、函数调用（工具调用）FunctionCalling让模型能够调用外部工具，来增强自身能力。模型会通过响应告诉你要调用哪个函数，传递什么参数，进而在智能体（或代码中），进行调用。MCP即：ModelContextProtocol:模型上下文协议。FunctionCalling是自已写函数，自已实现功能，MCP是通过本地服务或远程服务，将要实现的函数功能进行了封装，不需要自已再写。如果是企业内部的特写功能，也可以通过MCP服务来发布AI的函数。MCP协议不仅限于AI的使用，也可以作为一个标准的服务接口，与RESTful没有区别这种调用。以下展示了在CherryStudio中利用fetch进行网页内容获取，并交由大模型进行内容总结：以下展示了在CherryStudio中引用高德地图MCP服务进行天气查询和实时回复，先访问：https://lbs.a/，注册账号，实名认证，并获取APIKey后，进行以下配置：完成上述配置后，在对话中询问穿衣建议：四、RAG技术检索增强生成（Retrieval-augmentedGeneration），核心可以由三个部分构成：（1）检索：从向量数据库中检索与用户提问有关的内容片段。（2）增强：将检索出来的内容嵌入提示词中，为提示词增强内容。（3）生成：正常的大模型生成满足用户需求的回复内容。五、ReAct原理ReAct智能体是一种通过“推理与行动”（ReasoningandActing）框架，结合了思维链（ChainofThought,CoT）推理和外部工具使用的AI智能体。这一框架提升了大型语言模型（LLM）在处理复杂任务和决策时的能力，特别是在智能体工作流程中的应用。以下代码模拟了一个ReAct的工作流程，但是并未接入大模型，所以不是真实的ReAct：11234567891011121314151617181920importredefcalculator(expr:str)->str:"""安全且极简的计算工具：只允许数字、空格和*/ifnotre.fullmatch(r"[0-9+\-*/().\s]+",expr):return"ERROR:计算表达式不正确"try:returnstr(eval(expr))exceptExceptionase:returnf"ERROR:{e}"tools"calculator"calculator} 定义可用的工具列表classSimpleReActAgent:definit(self):self.scratchpad=[]defrun(self,question:str):print(f"Question:{question}\n")212122232425262728#这里用“脚本化策略”模拟LLM在ReAct中的推理/行动/反思过程steps=[{"thought":"先计算这个表达式。我先试试不加括号。","action":("calculator","5+3*2"),},{"thought":"结果是11，但题目有括号(5+3)，我忽略了优先级。需要反思并修正。",2930313233343536373839404142434445464748495051525354555657585960"reflection":"我刚才的行动不符合题意，应该先算括号，再乘2。","action":("calculator","(5+3)*2"),},]fori,stepinenumerate(steps,1):print(f"Thought{i}:{step['thought']}")if"reflection"instep:print(f"Reflection{i}:{step['reflection']}")tool_name,tool_input=step["action"]print(f"Action{i}:{tool_name}[{tool_input}]")observation=tools[tool_name](tool_input)print(f"Observation{i}:{observation}\n")self.scratchpad.append({"thought":step["thought"],"reflection":step.get("reflection"),"action":f"{tool_name}[{tool_input}]","observation":observation,})final_answer=self.scratchpad[-1]["observation"]print(f"FinalAnswer:{final_answer}")ifname =="main":q"请计算：(53)*2"agent=SimpleReActAgent()agent.run(q)那么如何通过对接大模型，来实现真实功能的ReAct呢，就需要优化提示词，接入大模型，并实现更加真实的功能了，下节课继续。总结，构成一个AI智能体的核心要素有哪些？函数调用、RAG技术、ReAct、提示词、应用开发。。。。对接云端AI大模型一、对接通义千问1、申请APIKey首先需要访问：/注册一个账号并进行实名认证。然后进入到大模型服务平台百炼的产品控制台中。并创建一个APIKey用于在Python调用其接口时进行鉴权。2、查看帮助文档当创建完APIKey后，则可以进入API参考页面，查看各种AI模型的调用，阿里云的文档非常完善，并且提供了针对各种编程语言的示例代码供用户参考。3、使用OpenAI进行调用由于OpenAI是最早提供大模型API对接的厂商，所以后续的一些AI厂商均提供了兼容OpenAI的SDK（开发工具包）接口的对接方案。在Python中需要使用“pipinstallopenai”安装SDK，然后通过以下代码完成对接。1 1 fromopenaiimportOpenAI2 importjson34 client=OpenAI(5 api_key="sk-30a2d87612584e2697867a796b3d4f01",6 base_url="/compatible-mode/v1",7 )89 completion=pletions.create(#模型列表：/zh/model-studio/getting-started/modelsmodel="qwen-plus",messages=[{'role':'system','content':'Youareahelpfulassistant.'},{'role''user''content''你是谁？'}],stream=True,stream_options={"include_usage":True}16 )17在流式响应中查看JSON的数据结构，便于准确获取响应正文#forchunkincompletion:# print(chunk.model_dump_json())21输出响应正文（根据JSON数据结构按层次输出即可）forchunkincompletion:iflen(chunk.choices)>0:print(chunk.choices[0].delta.content,end="")4、使用dotenv配置APIKey上述代码中的APIKey一般情况下来说，不建议直接写在代码中，通常的做法是将其保存到项目的环境变量文件中，然后通过dotenv库进行调用（本质就是配置为环境变量，与操作系统的环境变量读取是完全相同的操作），这样便于统一维护APIKey，同时不再暴露其Key到代码中。操作步骤为：（1）先利用pip安装dotenv第三方库：pipinstalldotenv（2）在当前项目的根目录下创建一个文件：.env（注意文件名以.开头），内容为：1 1 Dashscope_API_Key='sk-30a2d87612584e2697867a796b3d4f01'2 DeepSeek_API_Key='xxxxxxxxxxxxxxx'然后在代码中，可以直接通过dotenv的load_dotenv将其内容加载为环境变量，并进行读取。fromfromdotenvimportload_dotenvload_dotenv() 调用load_dotenv函数将.env的内容加载为环境变量client=OpenAI(api_key=os.getenv("Dashscope_API_Key"), 读取环境变量以获取APIKeybase_url="/compatible-mode/v1",)fromopenaiimportOpenAIimportjson,os123456789105、使用Dashscope进行调用通义千问模型除了兼容OpenAI外，阿里云本身也提供了SDK进行调用，只需要在Python中利用命令“pipinstalldashscope”进行安装即可直接使用，示例代码如下：112345678910111213141516171819importos,dashscopefromdotenvimportload_dotenvload_dotenv()messages=[{'role':'system','content':'youareahelpfulassistant'},{'role':'user','content':'你是谁？'}]responses=dashscope.Generation.call(api_key=os.getenv('Dashscope_API_Key'),model="qwen-plus",messages=messages,result_format='message',stream=True,incremental_output=True)forresponseinresponses:print(response.output.choices[0].message.content,end="")6、使用Requests进行调用几乎所有的AI模型厂商均提供了标准的HTTP协议的接口，所以即使没有提供SDK，也不影响在Python或其他编程语言中与AI模型进行交互。以下代码展示了如何利用Python的requests库通过发送原始HTTP请求的方式与模型进行通信。112345678910111213141516171819importrequests,jsonfromdotenvimportload_dotenvload_dotenv()url="/compatible-mode/v1/chat/completions"data={"model":"qwen-plus","messages":[{"role":"system","content":"Youareahelpfulassistant."},{"role":"user","content":"你是谁？"}],"stream":True}header={"Authorization":f"Bearer{os.getenv('Dashscope_API_Key')}"}response=requests.post(url=url,headers=header,json=data,stream=True)202122#输出格式类似于以下内容，所以需要特殊处理#b'data:{"choices":[{"delta":{"content":"","role":"assistant"},"}'#b''#b'data:{"choices":[{"finish_reason":null,"delta":{"content":"\xe6\xaf"},}'forchunkinresponse.iter_lines():232324252627chunk_list=chunk.decode('utf-8').split("data:")forlineinchunk_list:iflen(line)>10:chunk_json=json.loads(line)print(chunk_json['choices'][0]['delta']['content'],end='')上述代码相比较于SDK来说，处理过程会相对复杂一点，但是通用性会更强。Dashscope或OpenAI并没有为所有编程语言开发SDK，但是任何编程语言目前都支持HTTP协议的处理。二、对接DeepSeekDeepSeek也是是目前非常主流的AI大模型，无论是在编程还是对话及推理领域，均有显著优势，所以在进行各类AI应用或智能体开发时，对接DeepSeek模型也是很有必要的，只需要在DeepSeek官网注册一个账号，并进行实名认证后，开通APIKey即可使用。访问DeepSeek的API开放平台：/usage，并注册账号，然后申请开通一个APIKey。申请成功后，请务必要将明文的APIKey复制，并保存起来。否则，下一次需要重新创建。申请成功APIKey后，便可以在Python中进行调用了。Deepseek同样支持基于OpenAI的兼容模式调用，或直接使用HTTP协议进行调用，基础的调用代码如下。1123456789101112131415161718fromopenaiimportOpenAIimportosfromdotenvimportload_dotenvload_dotenv()client=OpenAI(api_key=os.getenv("DeepSeek_API_Key"),base_url="")completion=pletions.create(model="deepseek-chat",messages=[{"role":"system","content":"Youareahelpfulassistant"},{"role":"user","content":"你好，你是谁？"},],stream=True)forchunkincompletion:1919print(chunk.choices[0].delta.content,end="")其调用过程与调用通义前问基本一致，此处不再做过多解释。同时，通过Python的requests库也可以正常调用DeepSeek。urlurl="/chat/completions"data={"model":"deepseek-chat","messages":[{"role":"system","content":"Youareahelpfulassistant."},{"role":"user","content":"你好，你是谁？"}],"stream":True}header={"Authorization":f"Bearer{os.getenv('DeepSeek_API_Key')}"}response=requests.post(url=url,headers=header,json=data,stream=True)forchunkinresponse.iter_lines():chunk_list=chunk.decode('utf-8').split("data:")forlineinchunk_list:iflen(line)>10:chunk_json=json.loads(line)print(chunk_json['choices'][0]['delta']['content'],end='')importos,requests,jsonfromdotenvimportload_dotenvload_dotenv()123456789101112131415161718192021三、对接apimart.ai访问：https://apimart.ai/，使用Google或其他账号登录，并进行充值（支持微信和支付宝），申请APIKey等操作。apimart提供了不需要代理就可以正常访问GPT、Gemini等官方对国内IP不支持模型，且价格比官方价格还要优惠一些。使用Requests对接模型：load_dotenvload_dotenv()url="https://api.apimart.ai/v1/chat/completions"token=os.getenv("APIMart_API_Key")payload={"model""gpt-5.4-apimart", https://apimart.ai/model搜索模型ID"messages"{"role":"system","content":"YouareahelpfulAIassistant."},{"role""user""content""请简述AI人工智能的发展史，控制在500字以内。"}],"stream":False}headers={"Authorization":f"Bearer{token}","Content-Type":"application/json"}response=requests.post(url,json=payload,headers=headers)print(response.text)importrequests,osfromdotenvimportload_dotenv12345678910111213141516171819202122232425同样，也可以使用openai库进行调用：1 1 fromopenaiimportOpenAI2234567891011121314151617181920212223importrequests,osfromdotenvimportload_dotenvload_dotenv()url="https://api.apimart.ai/v1/"token=os.getenv("APIMart_API_Key")client=OpenAI(api_key=token,base_url=url)completion=pletions.create(model="gpt-5.4-apimart",messages=[{"role":"system","content":"Youareahelpfulassistant"},{"role":"user","content":"你好，你是谁？"},],stream=True)forchunkincompletion:ifchunk.choices[0].delta.contentisnotNone:print(chunk.choices[0].delta.content,end="")另外补充一些其他云端平台，如https://openrouter.ai/、ttps:///、或者是购买ttps:///四、ReAct实现以下代码实现了一个完整的ReAct过程，包含了真实的大模型接入、工具定义、工具调用、行动与反思，推理与结果处理等，同时，也优化了系统级提示词，让模型知道如何响应内容（这是模型推理的核心要素），代码则根据内容进行判断，实现完整ReAct。1123456789101112131415161718192021importjson,os,refromopenaiimportOpenAIfromdotenvimportload_dotenvload_dotenv()api_key=os.getenv("DeepSeek_API_Key")base_url=""client=OpenAI(api_key=api_key,base_url=base_url)defcalculator(expr):ifnotre.fullmatch(r"[0-9+\-*/().\s]+"expr):return"错误：表达式包含非法字符"try:returnstr(eval(expr))exceptExceptionase:returnf"错误：{e}"defread_file(file_path):withopen(file_path)asf:content=f.read()returncontent22tools={"calculator":calculator,"read_file":read_file,26 }2728defcall_llm(messages,temperature=0.2):completion=pletions.create(model="deepseek-chat",messages=messages,stream=False,temperature=temperature,35 )36result=completion.choices[0].message.contentprint(result)returnresult4041defparse_json(text:str):text=text.strip().replace("```json","").replace("```","").strip()returnjson.loads(text)4546defreact(question:str,max_steps=6):system_prompt="""你是一个ReAct智能体，需要进行“思考-行动-观察-反思”。\n50 ##可用工具：##\ncalculator：做数学计算，输入应是表达式字符串，例如"(5+3)*2"\nread_file\n5354 你必须只输出JSON（不要输出Markdown），格式如下：\n55 {"thought":当前思考","reflection":对上一步观察的反思与修正","action":{"tool":"calculator","input":"(5+3)*2"}null,"final_answer":"给用户的最终回答"或null60 }\n61 规则：\1)需要调用工具时：actionnull，final_answernull。2)可以直接回答时：actionnull，final_answernull。3)不要编造工具结果。4)thought和reflection尽量简短。66 """67messages=[{"role":"system","content":system_prompt},{"role":"user","content":question},71 ]7273 print(f"问题：{question}\n")74forstepinrange(1,max_steps+1):raw=call_llm(messages)try:obj=parse_json(raw)exceptExceptionase:messages+=[{"role":"assistant","content":raw},{"role""user""content"f"你的输出不是合法JSON（{e}），请只输出合法JSON。"},83 ]84 continue85thought=obj.get("thought","")reflection=obj.get("reflection","")action=obj.get("action")final_answer=obj.get("final_answer")90print(f"第{step}步思考：{thought}")print(f"第{step}{reflection}")9394iffinal_answer:95print(f"\n最终答案：{final_answer}")96returnfinal_answer9798ifnotaction:99messages+=[100{"role":"assistant","content":raw},101{"role":"user","content":"你必须给出action或final_answer之一。"},102]103continue104105106107108109110111112113114115tool_name=action.get("tool")tool_input=action.get("input","")iftool_namenotintools:observation=f"错误：未知工具{tool_name}"else:observation=tools[tool_name](tool_input)print(f"第{step}步行动：{tool_name}[{tool_input}]")print(f"第{step}步观察：{observation}\n")messages+=[116{"role":"assistant","content":raw},117{"role""user""content"f"工具观察结果：{observation}"},118]119120print("\n最终答案：超过最大步数，未完成任务。")121return"超过最大步数，未完成任务。"122123124ifname =="main":125#react("请计算：(8+7)*3。")126react("读取E:/Test.txt的文件内容。")上述代码主要讲解ReAct的核心原理，理解原理后，后续将主要使用ReAct框架如LangChain等进行操作，更加便捷。同时，我们平时所使用的各类AI智能体如编程助手、OpenClaw等，均使用了ReAct技术手段。1 1 问题：请计算：22--=(8+7)*3。2第第1'22--=(8+7)*3'(8+7)*3的结果。先计算这个表达式。第1步反思：第1步行动：calculator[(8+7)*3]第1步观察：4578910111213第2(8+7)*345'22--='可能是个笔误或多余字符。问题可能是'22-(8+7)*3'或类似。但原句是'22--=(8+7)*3'，这不合语法。或许用户想表达'22-(8+7)*3'？先计算22-45。第2步反思：用户输入不清晰，但基于常见错误，假设是减法。第2步行动：calculator[22-45]第2步观察：-23141516第3步思考：计算得到2245-23。用户原始输入'22--=(8+7)*3'可能包含作为'减号'的误输入。最终答案可能是-23。第3步反思：基于假设和计算，提供答案并说明假设。最终答案：根据计算，(8+7)*345，然后2245-23。用户输入'22--=(8+7)*3'可能意图是'22-(8+7)*3'，因此结果是-23。ToolCalling与MCP一、ToolCalling1、大模型的输入：文本，输出：文本，那么，如何让AI具备调用函数的能力？ToolCalling让模型能够调用外部工具，来增强自身能力。模型会通过响应告诉你要调用哪个函数，传递什么参数，进而在智能体（或代码中），进行调用。如何做到函数调用的？用户提交提示词时，需要额外增加一个参数用户告诉大模型函数的定义和功能，大模型根据用户的提示词来决定是否要调用某个函数传递某个参数，进而智能体通过解析大模型响应中的JSON字符串，来决定要不要调用。Python代码：importimportrequests,jsonfromopenaiimportOpenAI345678910111213141516system_prompt="""你是一名AI助手，具备函数调用的能力，但是如果提供的信息已经足够回答用户的问题，则不需要再进行函数调用。同时，请严格按照函数调用的方式进行处理，如果用户未提供函数所需参数，则必须询问，而不能自作主张。"""functions=[{"type":"function","function":{"name""get_weather", 函数名称"description":"查询某个城市的天气","parameters":{ #定义get_weather这个函数的参数"type":"object",17181920这个字",2122232425 }26 },2728 {

"properties":{"city" 第一个参数"type":"string","description":"中国的城市名称，不能在城市名称后面加上市}},"required":["city"]},293031323334353637383940414243444546474849 ]50

"type":"function","function":{"name":"write_file","description":"往某个文件中写入文本内容","parameters":{"type":"object","properties":{"filepath":{"type":"string","description""待写入文件的路径",},"content":{"type":"string","description":"文件写入的内容",}},"required":["filepath","content"]},}},实现函数get_weatherdefget_weather(city):apiUrl='/simpleWeather/query'54 apiKey='714eae911df6b6ebe436b7fc91489a88'55 requestParams={'key':apiKey,'city':city}5657 response=requests.post(apiUrl,params=requestParams)58ifresponse.status_code==200:returnresponse.json()else:return"Error"63实现文件内容写入defwrite_file(filepath,content):withopen(filepath,'w',encoding="utf-8")asfile:file.write(content)returnf"文件：{filepath}写入成功"6970defsend_messages(messages):client=OpenAI(api_key="sk-5f11c654fe144d9e8a2e8ca7f1689fea",base_url="")response=pletions.create(model="deepseek-chat",messages=messages,tools=functions,stream=False78 )79 returnresponse.choices[0].message80definvoke_function(func_name,**func_params):func=globals()[func_name]result=func(**func_params)returnresult8586ifname =='main':content明天在成都我应该穿什么衣服？"content请在D:/AITest/Test2.txt文件中写入：你好，蜗牛"content"你是谁？"messages=[{"role":"system","content":system_prompt},{"role":"user","content":content}]output=send_messages(messages)94ifoutput.tool_calls:func_name=output.tool_calls[0].func_args=eval(output.tool_calls[0].function.arguments)func_response=globals()[func_name](**func_args)print(func_response)100messages=[{"role":"user","content":f"请基于该信息：{func_response}\n来回答以下问题，请直接回答问题，不需要再进行函数调用，直接返回结果\n{content}"}]message=send_messages(messages)print(message.content)else:print("模型的回复:"output.content)二、MCP1、ModelContextProtocol:模型上下文协议FunctionCalling是自已写函数，自已实现功能，MCP是通过本地服务或远程服务，将要实现的函数功能进行了封装，不需要自已再写。如果是企业内部的特写功能，也可以通过MCP服务来发布AI的函数。所有云服务或企业内部的各类公共服务接口，均可以再使用MCP重新实现一遍。调用MCP服务的过程是由MCPClient完成。MCP协议不仅限于AI的使用，也可以作为一个标准的服务接口，与RESTful没有区别这种调用。中使用MCP，下载以下程序解压后放到：C:\Users\Denny.cherrystudio\binttps:///oven-sh/bun/releases/download/bun-v1.2.8/bun-windows-x64.zipttps:///astral-sh/uv/releases/download/0.6.12/uv-x86_64-pc-windows-msvc.zipMCP的服务类型有两种：STDIO：本地服务调用SSE：远程服务调用2、配置fetch用于提取网页内容3、配置file-system的MCP服务4、MySQL-mcp-server先安装：pipinstallmysql_mcp_server(建议3.11以上的版本)，官方网址：/desigcomputer/mysql_mcp_server5、高德地图服务调用效果：6、在Python中定义MCP服务安装：pipinstallmcp然后编写代码：1123456789frommcp.server.fastmcpimportFastMCPimportos,requestsmcp=FastMCP()1011121314151617181920212223242526272829@mcp.tool()defwrite_file(filename,content):"""写文件内容"""withopen(f"D:/CherryStudio/{filename}","w",encoding="utf-8")asfile:file.write(content)returnf"文件{filename}写入成功"@mcp.tool()defget_weather(city):"""查询某个城市的天气"""apiUrl='/simpleWeather/query'apiKey='714eae911df6b6ebe436b7fc91489a88'requestParams={'key':apiKey,'city':city}response=requests.get(apiUrl,params=requestParams)ifresponse.status_code==200:returnresponse.textelse:return"Error"ifname =="main":mcp.run(transport='stdio') 本地调用mcp.settings.port=80003030mcp.run(transport='sse')#远程调用配置：在Python中开发MCP应用一、Python开发MCP服务器端先安装mcp库：pipinstallmcp##使用SSEmcp.run(transport='sse')@mcp.tool()defwrite_file(filename,content):"""在文件中写入文本内容"""withopen(f"D:/CherryStudio/{filename}","w",encoding="utf-8")asfile:file.write(content)returnf"文件{filename}写入成功"@mcp.tool()defget_weather(city):"""查询某个城市的天气"""apiUrl='/simpleWeather/query'apiKey='714eae911df6b6ebe436b7fc91489a88'requestParams={'key':apiKey,'city':city}response=requests.get(apiUrl,params=requestParams)ifresponse.status_code==200:returnresponse.textelse:return"Error"ifname =="main":mcp.run(transport='stdio') 使用stdio#mcp.settings.port8000 修改端口161718192021222324252627282930313233343536mcp=FastMCP()@mcp.tool()deflist_desktop_files():"""获取指定文件中的所有文件列表"""desktop_path"D:/CherryStudio"returnos.listdir(desktop_path)frommcp.server.fastmcpimportFastMCPimportos,requests123456789101112131415二、MCPInspector1、安装Node.js，确认npx命令可用2、安装或启动npx-y@modelcontextprotocol/inspectoruvrunserver.py3、浏览器访问：:6274三、异步编程1．什么是异步编程在现代编程世界中，异步编程是提升应用程序性能的重要手段之一。Python提供了强大的异步编程功能，通过async和await关键字，使得程序能够以更高效的方式处理并发任务。本书之所以选用FastAPI作为后端开发框架，除了开发非常高效之外，支持异步通信和高并发性能也是其主要原因之一，因此FastAPI非常适用用于构建大型的应用程序。异步编程使得程序可以在等待某种操作时（例如网络请求或文件读取）执行其他操作，而不是单纯地阻塞住。这样可以提高程序的响应速度和资源利用率。在Python中，异步编程通常通过三个基本构件来实现：async：通常放置函数定义之前，用于定义一个异步函数。await：等待异步操作完成。asyncio：提供任务调度和事件循环的支持。2．异步代码演示为了更加深刻理解异步和同步的区别，以下代码模拟了一个常规的任务操作，每个任务完成的时间不同，当4个任务完成后，来看看一共需要多少时间。1123456789importtime,randomdefprocess_task(item):print("开始处理：",item)#假设一个任务的处理时间在0.5~2秒之间的一个随机数process_time=random.uniform(0.5,2.0)time.sleep(process_time)print(f"处理完成：{item}{process_time:.2f}秒")1010111213141516171819defprocess_all():items任务"任务"任务"任务D"]foriteminitems:process_task(item)ifname =="main":start=time.time()process_all()end=time.time()print(f"总耗时：{endstart:.2f}秒")上述代码是一个典型的同步代码，也就是任务是一个一个串行处理的，只有处理完任务A，才会开始处理任务B。代码输出结果如下：1开始处理：任务A2处理完成：任务A，耗时1.94秒3开始处理：任务B4处理完成：任务B，耗时0.53秒5开始处理：任务C6处理完成：任务C，耗时1.18秒7开始处理：任务D8处理完成：任务D，耗时1.37秒9总耗时：5.02秒由此可知，在处理任务A时，任务BCD是没有开始的，此时系统资源只用于处理任务A。但是如果系统资源本身是足够的，那么为何不在处理任务A的同时，也继续处理任务B、任务C呢？学习过多线程的读者可以知道，此时使用多线程也可以实现该目的。但是多线程和异步（或协程）在使用场景上是有区别的，多线程通常适用于CPU密集型，同时调度多个CPU来进行并发任务。而异步编程则不需要依赖于线程，相当于在一个线程中通过事件循环的非阻塞机制便可以实现并发任务，效率更高，适用于I/O密集型任务，且相较于线程，在Python中的实现更加简单，各类冲突问题也较少。以下代码展示了如何在Python中进行异步编程。11234567891011121314151617181920212223importtime,random,asyncioasyncdefprocess_task(item):print("开始处理："item)#假设一个任务的处理时间在0.5~2秒之间的一个随机数process_time=random.uniform(0.5,2.0)awaitasyncio.sleep(process_time)print(f"处理完成：{item}，耗时{process_time:.2f}秒")asyncdefprocess_all():items=["任务A","任务B","任务C","任务D"]tasks=[asyncio.create_task(process_task(item))foriteminitems]awaitasyncio.gather(*tasks)asyncdefmain():start=time.time()awaitprocess_all()end=time.time()print(f"总耗时：{end-start:.2f}秒")ifname =="main":asyncio.run(main())上述代码中使用async来定义函数，当要调用异步函数时，必须使用await关键字等待异步处理结束，同时在运行时不能再使用简单的函数调用的方式，而必须使用asyncio来进行调用和运行。代码运行结果如下。ABCD处理完成：任务D1.01秒处理完成：任务B1.36秒处理完成：任务C1.88秒处理完成：任务A1.91秒9 总耗时：1.92秒四、Python开发MCP客户端1、获取工具列表112345678importjson,os,asynciofrommcpimportClientSessionfrommcp.client.sseimportsse_clientfromdotenvimportload_dotenvload_dotenv()9101112131415161718192021222324252627282930313233server_url="/sse?key=46644f43e3a56ce81f6f3633c5994c67"api_key=os.getenv("DeepSeek_API_Key")classMCPClient:definit(self):self.sessions=[]存储服务器的会话及其上下文对象#初始化SSE服务器的连接，并获取可用工具列表。asyncdefinit_session(self):#创建SSE客户端并进入上下文streams_context=sse_client(url=server_url)streams=awaitstreams_context.aenter()session_context=ClientSession(*streams)session=awaitsession_context.aenter()awaitsession.initialize()#存储会话及其上下文self.sessions=[session,session_context,streams_context]#获取工具列表并建立映射response=awaitsession.list_tools()fortoolinresponse.tools:print(tool)#也可以直接调用某个工具responseawaitsession.call_tool("maps_weather",{"city":成都"})34343536#print(response)37383940414243444546474849505152535455565758responseawaitsession.call_tool("maps_geo",{"address":孵化园","city":"成都"})#print(response)#清理所有会话和连接资源，确保无资源泄漏。asyncdefcleanup(self):#清空上下文和会话等对象，并断开与服务器的连接session,session_context,streams_context=self.sessionstry:awaitsession_context.aexit(None,None,None)awaitstreams_context.aexit(None,None,None)awaitsession.aexit(None,None,None)except:passprint("所有会话已清理。")#异步调用主函数asyncdefmain():client=MCPClient()awaitclient.init_session()awaitclient.cleanup()初始化并输出工具列表#调用后必须清理ifname =='main':asyncio.run(main())2、AI整合与调用importimportjson,os,asynciofrommcpimportClientSessionfrommcp.client.sseimportsse_clientfromdotenvimportload_dotenvfromopenaiimportAsyncOpenAIload_dotenv()78910111213141516171819202122232425server_url="/sse?key=46644f43e3a56ce81f6f3633c5994c67"base_url=""api_key=os.getenv("DeepSeek_API_Key")classMCPClient:definit(self):self.sessions=[]存储服务器的会话及其上下文对象self.client=AsyncOpenAI(base_url=base_url,api_key=api_key)#初始化SSE服务器的连接，并获取可用工具列表。asyncdefinit_session(self):#############代码略#清理所有会话和连接资源，确保无资源泄漏。asyncdefcleanup(self):#############代码略处理用户的自然语言查询，通过工具调用完成任务并返回结果。asyncdefprocess_query(self,query:str):初始化消息列表messages=[{"role":"user","content":query}]30available_tools 收集所有可用工具session,_,_self.sessions 获取session对象response=awaitsession.list_tools()fortoolinresponse.tools:构造满足FunctionCalling规范的函数声明function={"type":"function","function":{"name":,"description":tool.description,"parameters":tool.inputSchema,42 },43 }44 available_tools.append(function)45向模型发送初始请求（带tools参数，非流式）response=pletions.create(model="deepseek-chat",messages=messages,tools=available_tools,51 )52读取模型的响应message=response.choices[0].message55明确了大模型返回的函数调用及参数值后，开始进行工具调用ifmessage.tool_calls:一次可能存在多个函数调用的情况fortoolinmessage.tool_calls:function_name=function_args=json.loads(tool.function.arguments)try:result=awaitsession.call_tool(function_name,function_args)print(f"[{function_name}参数{function_args}]，结果：{result}")将工具的输出结果添加到消息体中messages.append({"role":"user","content":str(result.content)})exceptExceptionase:print(f"{function_name}出错：{str(e)}")6970 print(messages)71再次将MCP返回的结果添加给大模型进行二次请求response=pletions.create(model="deepseek-chat",messages=messages,stream=True77 )asyncforchunkinresponse:text=chunk.choices[0].delta.contentiftext:818182838485868788899091929394959697yieldtext 异步生成器方式#异步调用主函数asyncdefmain():client=MCPClient()awaitclient.init_session() 初始化并输出工具列表cess_query("查询成都市环球中心附近酒店。"):print(text,end='')awaitclient.cleanup()调用后必须清理ifname =='main':asyncio.run(main())以上是针对单个SSE服务器情况的代码，MCP客户端也可以连接到多台MCP服务器，则需要将其进行区分，可以将连接到的不同的服务器用self.sessions来进行定义，设置self.sessions=[]，再添加不同客户端实例即可，保存到不同的Session会话中。在调用时，需要针对大模型返回的函数来进行判断，确定调用哪个Session下的函数。RAG检索增强生成技术一、概述大模型是一种厂商预训练模型，也就是说，在与大模型进行AI对话的过程中所得到的结果，其实是大模型训练过程中获取到的知识。所以为了获取到更多的知识，通常在设计AI应用时，为其增加了联网搜索的功能。但是联网搜索的功能，并不能获取到所有知识，依然是一些公开在网络上的信息而已。还有大量的内容，是属于私有的，无法在网络上找到答案的。比如一些行业的内部资料，一家公司的规章制度，或者是一次会议的会议纪要等等，各种各样的资料汇聚成了私有知识。那么，如何让大模型也有机会学习到这些私有化的知识，并且可以对其进行整理和总结，进而在AI对话中体现出来呢？通常，要让大模型学习私有知识，一般有三种方案：（1）将私有知识进行整理后，通过模型微调手段将新知识注入到模型中，其核心就是改变模型的权重，使其具备理解新知识的能力。该方案比较适用于基于一个现有的模型来构建某个行业或领域的专有模型，如医疗大模型、法律大模型、金融大模型等，相对来说，效果较好，但是训练成本较高。（2）基于提示词来学习私有化知识，是指当用户在进行提问时，如果需要用到一些私有知识，则将该私有知识作为提示词的一部分交由大模型理解。再在提示词在添加对于该知识的提问，大模型就可以学习到私有知识并进行回答。AI智慧课堂的项目中，基于课程内容生成课堂笔记或考题等操作，便属于这一方案。或者某些AI对话程序可以让用户上传一份文档，再基于该文档进行提问，也是属于这一类。该方案无技术门槛，操作简单，但是对于模型的上下文大小有所要求，如果能够支持的上下文长度太小，则无法将大量的文档通过提示词进行学习。同时，由于需要进行大量的文档嵌入操作，将显著增加模型的运算和响应时间。再有，如果用户是使用的云端模型，则通常还存在Token消耗的成本。（3）基于RAG（检索增强生成，Retrieval-AugmentedGeneration）技术来获取私有知识。该技术方案将综合前面两种方案的优缺点，其核心思路是将私有知识分段（也叫切片、分片、分块，本书一律采用分段来描述）后保存到向量数据库中，再根据用户的提问来检索出相关的段落，作为提示词的一部分交由大模型学习和总结。该方案不需要进行模型微调，无训练成本，同时由于只检索了相关段落交由大模型，所以对大模型的上下文大小无要求，且Token消耗较低，是综合了各方面优势的一种平衡方案。但是也存在一个很大的缺点，那就是准确率有所下降，对输入的私有知识无全局理解能力。二、RAG工作机制在阐述本项目的功能与实现思路之前，先就前文提到的RAG与向量数据库两个概念进行解释说明。首先，RAG与向量数据库是两个独立的技术，各自有自己的场景和价值。只是二者经常会放到一起解决私有知识库的处理问题，所以本节内容一并对两个技术的工作原理做一下解释。RAG的全称是检索增强生成，由三个词语构成，分别对应了RAG的三个重要的工作流程。检索是指根据用户的提问，从大量分段后的文档中检索出相关的文档片段。增强是指将检索出的内容作为大模型提示词的一部分来增强大模型知识的能力。生成则是大模型将新知识生成给用户。这一过程最难的地方就在于，如何根据用户的提问来检索出相似度最高的文档，以嵌入到大模型提示词中。所以，计算文档之间的相似度就成为了一个必须要解决的问题。1、RAG工作流程（1）第一步：将文档进行切片（或叫分段），可以按照章节、段落、字数等进行分段。（2）第二步：将切片段落的内容，转换为词嵌入向量，并保存到向量数据库中（句向量，无论分段内容有多长，向量维度固定）。（3）第三步：对向量进行存储并建立索引，以进行快速检索。（4）第四步：将用户的提问转换为相同维度的向量，并与向量数据库中的向量进行相似度检索。（5）第五步：将检索到的相似切片注入到提示词中，并交由大模型进行回复。2．理解句向量要计算用户提问与文档片段之间的相似度，首先就需要将文本向量化，即将文字变成N维的数字，这一过程称为词嵌入。在AI模型中，针对文本的训练和推理，首先就需要将文本向量化，以下代码演示了如何通过Ollama的词嵌入接口获取到某句话的词向量。client=Client(host="89:11434")client=Client(host="89:11434")result=client.embed(model="qwen3-embedding:8b",input="今天的天气真不错！")print(result.embeddings) #输出向量值print(len(result.embeddings[0])) 输出向量维度，4096#通过HTTP协议获取句向量importrequestsdata={"model":"qwen3-embedding:8b","input":"今天的天气真不错！"}resp=requests.post("89:11434/api/embed",json=data)print(resp.json()['embeddings'])使用SDK获取某个段落的句向量fromollamaimportClient1234567891011121314大致输出的内容为：1[[0.0032885247,-0.020798128,0.023944877,0.0012922653,0.022558969,...]]240963[[0.0032885247,-0.020798128,0.023944877,0.0012922653,0.022558969,...]]可以看出，每条语句的向量有4096个维度（如果只是一个Token，也是这样的维度，如果是很长的一句话，也是这个向量维度）。同时，不同的模型，向量的维度是不同的。这里的4096仅代表“qwen3-embedding:8b”这个模型的向量维度为4096。另外，这个维度的含义就是表示用了4096个数字来描述一句话（也可以称为句向量），这里面的数字均是模型训练后的结果。3、使用OpenAI进行词嵌入#3072#3072print(len(embedding))print(embedding)embeddinggen_embedding("今天的天气真不错！")api_key=os.getenv("API_Mart_Key")base_url="https://api.apimar