2 智能体产品设计隐私泄露歧视、危害国家安全。设计时根据诉求发掘深层的意图合规诉求。展示可以通过 分步骤、需要对比的需要左中右分栏。代码生成 智能体分开发者、review视角。语音交互设计语音用户界面VUI设计原则1. 首轮简短层级递进。 减少用户负担。2. 听觉建议 传达时先传达结论再传达细节。3. 关键操作信息复述并让客户确认。4. 与视觉屏幕 触摸屏触觉结合避免 嘈杂环境下不可操作。输出1. 通过 首Token降低用户等待焦虑。2. 工具调用等给出中间结果。避免用户以为没有响应。打断1. 放弃TTS继承上下文立即开始新的意图理解。播报 断点续传机制。 继续播放时可以选择段落开头或者继续播放。人机协同型智能体1. 中心化编排先拆分再 汇总。2. 流水线结构3. 循环模式。智能体触发 智能体之间的交互规范中间环节、报错的可见性。AG-UIAgent-User Interaction Protocol智能体—用户交互协议是一个开放、轻量、事件驱动的协议用来标准化「后端 AI 智能体」与「前端用户界面」之间的实时双向通信。由 CopilotKit 团队于 2025 年 5 月开源。解决的痛点很具体过去做一个 Agent 应用前端要自己用 WebSocket/SSE 手搓一套流式协议——怎么推 token、怎么显示正在调用工具、怎么同步 Agent 内部状态、怎么让用户中途打断或补充信息。Agent 通过标准 HTTP/SSE也支持 WebSocket、Webhook、二进制通道向前端推送一个 JSON 事件序列约 16 种标准事件类型主要分四类消息事件文本增量流式输出工具调用事件工具名、参数、结果状态补丁事件用 delta 增量同步 Agent 状态而不是全量重传生命周期事件run 开始/结束/报错3.1 智能体高代码开发高代码允许 硬件级资源优化允许自定义微服务粒度、通信协议、增加数据分片策略。能控制技术债务。一个节点 同时完成很多任务的时候成功率会迅速降低。 plan的长度越长准确率也越低。设计项目时1. 模块化分层智能体系统应采用感知层、决策层、执行层的三层架构。 路由层、2. 分层记忆机制4层工作、情景、语义、元信息、记忆权限隔离、保留周期控制。* 安全控制设计实现输入过滤防指令注入、权限校验API 白名单、输出审查。接口统一API规范、负载均衡。对话历史内容过长的四种解决办法转写、检索、压缩、隔离。react模式的开发对模型的要求高研发成本高。 所以生产环境用工作流 速度最快、最稳定。工作流的优化1. 串行任务 并行化。 2. 模型微调 或者换模型。 3. 流式返回。技术选型信息收集智能体3.5 function call模型不知道什么时候调用怎么调用。ds-3调用范例import json from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, ToolMessage from dotenv import load_dotenv import os import os import getpass # v2 版本只需填写从千帆平台获取的 API Key # 获取路径控制台 - 安全认证 - API Key - 创建API Key选择千帆ModelBuilder qianfan_api_key getpass.getpass(Please input QIANFAN API Key (格式: bce-v3/ALTAK-xxx/xxx):) os.environ[API_KEY] qianfan_api_key # 初始化大模型 llm ChatOpenAI( modeldeepseek-v3.2, temperature0, api_keyos.getenv(API_KEY), base_urlhttps://qianfan.baidubce.com/v2 ) print(大模型初始化成功!)# 定义一个简单的加法函数 tool def add(a: int, b: int) - int: 将两个数字相加。 Args: a: 第一个数字 b: 第二个数字 Returns: 两数之和 return a b # 将工具绑定到模型 llm_with_tools llm.bind_tools([add]) print(工具定义完成!) print(f工具名称: {add.name}) print(f工具描述: {add.description})# 测试函数调用 query 请帮我计算 123 加 456 等于多少? messages [HumanMessage(contentquery)] response llm_with_tools.invoke(messages) print( 模型响应 ) print(f是否调用工具: {bool(response.tool_calls)}) if response.tool_calls: tool_call response.tool_calls[0] print(f\n调用的工具: {tool_call[name]}) print(f工具参数: {tool_call[args]}) # 执行实际的函数调用 result add.invoke(tool_call[args]) print(f\n执行结果: {result}) 模型响应 是否调用工具: True调用的工具: add工具参数: {a: 123, b: 456}执行结果: 579def run_conversation(query: str, llm_with_tools, tools_dict: dict): 运行完整的对话流程 print(f用户提问: {query}\n) # 1. 发送用户消息 messages [HumanMessage(contentquery)] response llm_with_tools.invoke(messages) # 2. 检查是否需要调用工具 if not response.tool_calls: print(模型直接回复:, response.content) return response.content # 3. 执行工具调用 messages.append(response) for tool_call in response.tool_calls: tool_name tool_call[name].strip() tool_args tool_call[args] print(f调用工具: {tool_name}) print(f参数: {tool_args}) # 执行工具 tool_result tools_dict[tool_name].invoke(tool_args) print(f工具返回: {tool_result}\n) # 将工具结果添加到消息历史 messages.append( ToolMessage( contentstr(tool_result), tool_call_idtool_call[id] ) ) # 4. 获取最终回复 final_response llm_with_tools.invoke(messages) print(最终回复:, final_response.content) return final_response.content# 测试完整流程tools_dict {add: add}run_conversation(帮我算一下 789 321,llm_with_tools,tools_dict)tool def multiply(a: float, b: float) - float: 将两个数字相乘。 Args: a: 第一个数字 b: 第二个数字 return a * b tool def divide(a: float, b: float) - float: 将第一个数字除以第二个数字。 Args: a: 被除数 b: 除数 if b 0: return 错误: 除数不能为0 return a / b tool def power(base: float, exponent: float) - float: 计算幂运算。 Args: base: 底数 exponent: 指数 return base ** exponent # 绑定所有工具 calculator_tools [add, multiply, divide, power] llm_calculator llm.bind_tools(calculator_tools) tools_dict {t.name: t for t in calculator_tools} print(f已注册 {len(calculator_tools)} 个工具:) for t in calculator_tools: print(f - {t.name}: {t.description})tool def get_current_weather(location: str, unit: str celsius) - str: 获取指定城市的当前天气信息。 Args: location: 城市名称,例如 北京, 上海, 深圳 unit: 温度单位, celsius(摄氏度) 或 fahrenheit(华氏度) Returns: 天气信息的JSON字符串 # 模拟天气数据 weather_data { 北京: {temperature: 15, condition: 晴朗, humidity: 45}, 上海: {temperature: 22, condition: 多云, humidity: 65}, 深圳: {temperature: 28, condition: 小雨, humidity: 80}, 广州: {temperature: 26, condition: 阴天, humidity: 70}, } if location in weather_data: data weather_data[location].copy() if unit fahrenheit: data[temperature] data[temperature] * 9/5 32 data[unit] °F else: data[unit] °C return json.dumps(data, ensure_asciiFalse) else: return json.dumps({error: f未找到城市 {location} 的天气信息}, ensure_asciiFalse) tool def get_weather_forecast(location: str, days: int 3) - str: 获取指定城市未来几天的天气预报。 Args: location: 城市名称 days: 预报天数(1-7天) Returns: 天气预报的JSON字符串 # 模拟预报数据 forecast [] conditions [晴, 多云, 阴, 小雨] for i in range(min(days, 7)): forecast.append({ day: f第{i1}天, temperature_high: 25 i, temperature_low: 15 i, condition: conditions[i % len(conditions)] }) return json.dumps({ location: location, forecast: forecast }, ensure_asciiFalse) # 绑定天气工具 weather_tools [get_current_weather, get_weather_forecast] llm_weather llm.bind_tools(weather_tools) weather_tools_dict {t.name: t for t in weather_tools} print(天气助手工具已准备好!) # 测试天气查询 weather_queries [ # 北京现在天气怎么样?, # 上海和深圳哪个城市温度更高?, 告诉我广州未来3天的天气预报, ] for query in weather_queries: print(\n *60) run_conversation(query, llm_weather, weather_tools_dict) print(*60)# 测试并行调用 query 请告诉我北京、上海和深圳这三个城市现在的天气情况 print(f用户提问: {query}\n) messages [HumanMessage(contentquery)] response llm_weather.invoke(messages) if response.tool_calls: print(f模型决定并行调用 {len(response.tool_calls)} 个工具:\n) messages.append(response) # 执行所有工具调用 for i, tool_call in enumerate(response.tool_calls, 1): tool_name tool_call[name] tool_args tool_call[args] print(f[{i}] 调用 {tool_name}) print(f 参数: {tool_args}) result weather_tools_dict[tool_name].invoke(tool_args) print(f 结果: {result}\n) messages.append( ToolMessage( contentstr(result), tool_call_idtool_call[id] ) ) # 获取最终回复 final_response llm_weather.invoke(messages) print(\n最终回复:) print(final_response.content)# 测试天气查询 weather_queries [ # 北京现在天气怎么样?, # 上海和深圳哪个城市温度更高?, 告诉我广州未来3天的天气预报, ] for query in weather_queries: print(\n *60) run_conversation(query, llm_weather, weather_tools_dict) print(*60)用户提问: 告诉我广州未来3天的天气预报调用工具: get_weather_forecast参数: {location: 广州, days: 3}工具返回: {location: 广州, forecast: [{day: 第1天, temperature_high: 25, temperature_low: 15, condition: 晴}, {day: 第2天, temperature_high: 26, temperature_low: 16, condition: 多云}, {day: 第3天, temperature_high: 27, temperature_low: 17, condition: 阴}]}最终回复: 根据天气预报广州未来3天的天气情况如下**第1天**- 天气晴- 最高温度25°C- 最低温度15°C**第2天**- 天气多云- 最高温度26°C- 最低温度16°C**第3天**- 天气阴- 最高温度27°C- 最低温度17°C总体来说广州未来3天天气逐渐转暖温度在15-27°C之间天气从晴天逐渐转为多云和阴天。建议您根据天气变化适当调整衣物。# 测试并行调用 query 请告诉我北京、上海和深圳这三个城市现在的天气情况 print(f用户提问: {query}\n) messages [HumanMessage(contentquery)] response llm_weather.invoke(messages) if response.tool_calls: print(f模型决定并行调用 {len(response.tool_calls)} 个工具:\n) messages.append(response) # 执行所有工具调用 for i, tool_call in enumerate(response.tool_calls, 1): tool_name tool_call[name] tool_args tool_call[args] print(f[{i}] 调用 {tool_name}) print(f 参数: {tool_args}) result weather_tools_dict[tool_name].invoke(tool_args) print(f 结果: {result}\n) messages.append( ToolMessage( contentstr(result), tool_call_idtool_call[id] ) ) # 获取最终回复 final_response llm_weather.invoke(messages) print(\n最终回复:) print(final_response.content)3.7 大模型工具调用mcpmcp实质是带服务发现的function call。 模型与工具的适配原来m*n的适配成本。现在是n个。工具需要自主反思的机制。SKILLS渐进式披露 只看每一个的描述来决定是否选用。 选用的技能才加载全部内容。 最后执行时才去调用。环境安装pip install fastmcp[all]mcp服务1. tool:2. 提示词模板 一些环境信息3. 资源的定义4. 辅助函数服务端需要常驻运行python mcp_server.py客户端 定义连接服务器的接口包括1. 建立连接调用工具# 便捷方法 async def multiply(self, a, b): return await self.call(multiply, aa, bb) async def calculate(self, operation, a, b): return await self.call(calculate, operationoperation, aa, bb) async def analyze(self, text): return await self.call(text_analysis, texttext) async def time(self, formatreadable): return await self.call(get_current_time, formatformat) # 使用示例 if __name__ __main__: async def test(): client MCPClient() await client.connect() # 测试工具调用 result await client.multiply(7, 8) print(f7 * 8 {result}) result await client.calculate(add, 10, 20) print(f10 20 {result}) await client.disconnect() asyncio.run(test())一般skill都是自动生成的。skill生成包含以下几个东西1.通用智能体如何实现swarm架构的去中心化1. 任务判定2. 任务拆解任务依赖关系。3. while循环执行任务。调用具体的执行任务的函数。while i len(tasks)任务的执行管理用from swarm import Swarm, Agentswarm.run( agentmarketing_agent, messages[{role: user, content: 帮我策划Q4的市场活动做一个完整的规划。}], context_variablescontext )4.代码检查任务1. 生成代码2. 测试代码 一系列参数exec执行。3. 大模型修正代码。4. 循环2/3直到全部成功或达到最大步数。大模型调用函数1. 用大模型输出选择调用的函数然后用正则解析获取大模型想要调用的函数名。2. 然后用字符串判定为真的方法去调用函数。大模型生成代码的执行1. 检查代码2. 用python自带的可以将字符串变为 函数名。# 执行代码提取函数 exec(code, {}, local_vars)eval执行函数并获取值。code 2 2result eval(code)AI员工需要构建一个Workspace。一种尝试围绕Agent构建「团队研发仓库」的实践一个让团队所有人都可以沉淀为「Agent可用资产」的载体。 
长河君 · 2026/10/29 1:49:33 · 阅读 1.2k 【无标题】学习笔记
2 智能体产品设计隐私泄露歧视、危害国家安全。设计时根据诉求发掘深层的意图合规诉求。展示可以通过 分步骤、需要对比的需要左中右分栏。代码生成 智能体分开发者、review视角。语音交互设计语音用户界面VUI设计原则1. 首轮简短层级递进。 减少用户负担。2. 听觉建议 传达时先传达结论再传达细节。3. 关键操作信息复述并让客户确认。4. 与视觉屏幕 触摸屏触觉结合避免 嘈杂环境下不可操作。输出1. 通过 首Token降低用户等待焦虑。2. 工具调用等给出中间结果。避免用户以为没有响应。打断1. 放弃TTS继承上下文立即开始新的意图理解。播报 断点续传机制。 继续播放时可以选择段落开头或者继续播放。人机协同型智能体1. 中心化编排先拆分再 汇总。2. 流水线结构3. 循环模式。智能体触发 智能体之间的交互规范中间环节、报错的可见性。AG-UIAgent-User Interaction Protocol智能体—用户交互协议是一个开放、轻量、事件驱动的协议用来标准化「后端 AI 智能体」与「前端用户界面」之间的实时双向通信。由 CopilotKit 团队于 2025 年 5 月开源。解决的痛点很具体过去做一个 Agent 应用前端要自己用 WebSocket/SSE 手搓一套流式协议——怎么推 token、怎么显示正在调用工具、怎么同步 Agent 内部状态、怎么让用户中途打断或补充信息。Agent 通过标准 HTTP/SSE也支持 WebSocket、Webhook、二进制通道向前端推送一个 JSON 事件序列约 16 种标准事件类型主要分四类消息事件文本增量流式输出工具调用事件工具名、参数、结果状态补丁事件用 delta 增量同步 Agent 状态而不是全量重传生命周期事件run 开始/结束/报错3.1 智能体高代码开发高代码允许 硬件级资源优化允许自定义微服务粒度、通信协议、增加数据分片策略。能控制技术债务。一个节点 同时完成很多任务的时候成功率会迅速降低。 plan的长度越长准确率也越低。设计项目时1. 模块化分层智能体系统应采用感知层、决策层、执行层的三层架构。 路由层、2. 分层记忆机制4层工作、情景、语义、元信息、记忆权限隔离、保留周期控制。* 安全控制设计实现输入过滤防指令注入、权限校验API 白名单、输出审查。接口统一API规范、负载均衡。对话历史内容过长的四种解决办法转写、检索、压缩、隔离。react模式的开发对模型的要求高研发成本高。 所以生产环境用工作流 速度最快、最稳定。工作流的优化1. 串行任务 并行化。 2. 模型微调 或者换模型。 3. 流式返回。技术选型信息收集智能体3.5 function call模型不知道什么时候调用怎么调用。ds-3调用范例import json from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, ToolMessage from dotenv import load_dotenv import os import os import getpass # v2 版本只需填写从千帆平台获取的 API Key # 获取路径控制台 - 安全认证 - API Key - 创建API Key选择千帆ModelBuilder qianfan_api_key getpass.getpass(Please input QIANFAN API Key (格式: bce-v3/ALTAK-xxx/xxx):) os.environ[API_KEY] qianfan_api_key # 初始化大模型 llm ChatOpenAI( modeldeepseek-v3.2, temperature0, api_keyos.getenv(API_KEY), base_urlhttps://qianfan.baidubce.com/v2 ) print(大模型初始化成功!)# 定义一个简单的加法函数 tool def add(a: int, b: int) - int: 将两个数字相加。 Args: a: 第一个数字 b: 第二个数字 Returns: 两数之和 return a b # 将工具绑定到模型 llm_with_tools llm.bind_tools([add]) print(工具定义完成!) print(f工具名称: {add.name}) print(f工具描述: {add.description})# 测试函数调用 query 请帮我计算 123 加 456 等于多少? messages [HumanMessage(contentquery)] response llm_with_tools.invoke(messages) print( 模型响应 ) print(f是否调用工具: {bool(response.tool_calls)}) if response.tool_calls: tool_call response.tool_calls[0] print(f\n调用的工具: {tool_call[name]}) print(f工具参数: {tool_call[args]}) # 执行实际的函数调用 result add.invoke(tool_call[args]) print(f\n执行结果: {result}) 模型响应 是否调用工具: True调用的工具: add工具参数: {a: 123, b: 456}执行结果: 579def run_conversation(query: str, llm_with_tools, tools_dict: dict): 运行完整的对话流程 print(f用户提问: {query}\n) # 1. 发送用户消息 messages [HumanMessage(contentquery)] response llm_with_tools.invoke(messages) # 2. 检查是否需要调用工具 if not response.tool_calls: print(模型直接回复:, response.content) return response.content # 3. 执行工具调用 messages.append(response) for tool_call in response.tool_calls: tool_name tool_call[name].strip() tool_args tool_call[args] print(f调用工具: {tool_name}) print(f参数: {tool_args}) # 执行工具 tool_result tools_dict[tool_name].invoke(tool_args) print(f工具返回: {tool_result}\n) # 将工具结果添加到消息历史 messages.append( ToolMessage( contentstr(tool_result), tool_call_idtool_call[id] ) ) # 4. 获取最终回复 final_response llm_with_tools.invoke(messages) print(最终回复:, final_response.content) return final_response.content# 测试完整流程tools_dict {add: add}run_conversation(帮我算一下 789 321,llm_with_tools,tools_dict)tool def multiply(a: float, b: float) - float: 将两个数字相乘。 Args: a: 第一个数字 b: 第二个数字 return a * b tool def divide(a: float, b: float) - float: 将第一个数字除以第二个数字。 Args: a: 被除数 b: 除数 if b 0: return 错误: 除数不能为0 return a / b tool def power(base: float, exponent: float) - float: 计算幂运算。 Args: base: 底数 exponent: 指数 return base ** exponent # 绑定所有工具 calculator_tools [add, multiply, divide, power] llm_calculator llm.bind_tools(calculator_tools) tools_dict {t.name: t for t in calculator_tools} print(f已注册 {len(calculator_tools)} 个工具:) for t in calculator_tools: print(f - {t.name}: {t.description})tool def get_current_weather(location: str, unit: str celsius) - str: 获取指定城市的当前天气信息。 Args: location: 城市名称,例如 北京, 上海, 深圳 unit: 温度单位, celsius(摄氏度) 或 fahrenheit(华氏度) Returns: 天气信息的JSON字符串 # 模拟天气数据 weather_data { 北京: {temperature: 15, condition: 晴朗, humidity: 45}, 上海: {temperature: 22, condition: 多云, humidity: 65}, 深圳: {temperature: 28, condition: 小雨, humidity: 80}, 广州: {temperature: 26, condition: 阴天, humidity: 70}, } if location in weather_data: data weather_data[location].copy() if unit fahrenheit: data[temperature] data[temperature] * 9/5 32 data[unit] °F else: data[unit] °C return json.dumps(data, ensure_asciiFalse) else: return json.dumps({error: f未找到城市 {location} 的天气信息}, ensure_asciiFalse) tool def get_weather_forecast(location: str, days: int 3) - str: 获取指定城市未来几天的天气预报。 Args: location: 城市名称 days: 预报天数(1-7天) Returns: 天气预报的JSON字符串 # 模拟预报数据 forecast [] conditions [晴, 多云, 阴, 小雨] for i in range(min(days, 7)): forecast.append({ day: f第{i1}天, temperature_high: 25 i, temperature_low: 15 i, condition: conditions[i % len(conditions)] }) return json.dumps({ location: location, forecast: forecast }, ensure_asciiFalse) # 绑定天气工具 weather_tools [get_current_weather, get_weather_forecast] llm_weather llm.bind_tools(weather_tools) weather_tools_dict {t.name: t for t in weather_tools} print(天气助手工具已准备好!) # 测试天气查询 weather_queries [ # 北京现在天气怎么样?, # 上海和深圳哪个城市温度更高?, 告诉我广州未来3天的天气预报, ] for query in weather_queries: print(\n *60) run_conversation(query, llm_weather, weather_tools_dict) print(*60)# 测试并行调用 query 请告诉我北京、上海和深圳这三个城市现在的天气情况 print(f用户提问: {query}\n) messages [HumanMessage(contentquery)] response llm_weather.invoke(messages) if response.tool_calls: print(f模型决定并行调用 {len(response.tool_calls)} 个工具:\n) messages.append(response) # 执行所有工具调用 for i, tool_call in enumerate(response.tool_calls, 1): tool_name tool_call[name] tool_args tool_call[args] print(f[{i}] 调用 {tool_name}) print(f 参数: {tool_args}) result weather_tools_dict[tool_name].invoke(tool_args) print(f 结果: {result}\n) messages.append( ToolMessage( contentstr(result), tool_call_idtool_call[id] ) ) # 获取最终回复 final_response llm_weather.invoke(messages) print(\n最终回复:) print(final_response.content)# 测试天气查询 weather_queries [ # 北京现在天气怎么样?, # 上海和深圳哪个城市温度更高?, 告诉我广州未来3天的天气预报, ] for query in weather_queries: print(\n *60) run_conversation(query, llm_weather, weather_tools_dict) print(*60)用户提问: 告诉我广州未来3天的天气预报调用工具: get_weather_forecast参数: {location: 广州, days: 3}工具返回: {location: 广州, forecast: [{day: 第1天, temperature_high: 25, temperature_low: 15, condition: 晴}, {day: 第2天, temperature_high: 26, temperature_low: 16, condition: 多云}, {day: 第3天, temperature_high: 27, temperature_low: 17, condition: 阴}]}最终回复: 根据天气预报广州未来3天的天气情况如下**第1天**- 天气晴- 最高温度25°C- 最低温度15°C**第2天**- 天气多云- 最高温度26°C- 最低温度16°C**第3天**- 天气阴- 最高温度27°C- 最低温度17°C总体来说广州未来3天天气逐渐转暖温度在15-27°C之间天气从晴天逐渐转为多云和阴天。建议您根据天气变化适当调整衣物。# 测试并行调用 query 请告诉我北京、上海和深圳这三个城市现在的天气情况 print(f用户提问: {query}\n) messages [HumanMessage(contentquery)] response llm_weather.invoke(messages) if response.tool_calls: print(f模型决定并行调用 {len(response.tool_calls)} 个工具:\n) messages.append(response) # 执行所有工具调用 for i, tool_call in enumerate(response.tool_calls, 1): tool_name tool_call[name] tool_args tool_call[args] print(f[{i}] 调用 {tool_name}) print(f 参数: {tool_args}) result weather_tools_dict[tool_name].invoke(tool_args) print(f 结果: {result}\n) messages.append( ToolMessage( contentstr(result), tool_call_idtool_call[id] ) ) # 获取最终回复 final_response llm_weather.invoke(messages) print(\n最终回复:) print(final_response.content)3.7 大模型工具调用mcpmcp实质是带服务发现的function call。 模型与工具的适配原来m*n的适配成本。现在是n个。工具需要自主反思的机制。SKILLS渐进式披露 只看每一个的描述来决定是否选用。 选用的技能才加载全部内容。 最后执行时才去调用。环境安装pip install fastmcp[all]mcp服务1. tool:2. 提示词模板 一些环境信息3. 资源的定义4. 辅助函数服务端需要常驻运行python mcp_server.py客户端 定义连接服务器的接口包括1. 建立连接调用工具# 便捷方法 async def multiply(self, a, b): return await self.call(multiply, aa, bb) async def calculate(self, operation, a, b): return await self.call(calculate, operationoperation, aa, bb) async def analyze(self, text): return await self.call(text_analysis, texttext) async def time(self, formatreadable): return await self.call(get_current_time, formatformat) # 使用示例 if __name__ __main__: async def test(): client MCPClient() await client.connect() # 测试工具调用 result await client.multiply(7, 8) print(f7 * 8 {result}) result await client.calculate(add, 10, 20) print(f10 20 {result}) await client.disconnect() asyncio.run(test())一般skill都是自动生成的。skill生成包含以下几个东西1.通用智能体如何实现swarm架构的去中心化1. 任务判定2. 任务拆解任务依赖关系。3. while循环执行任务。调用具体的执行任务的函数。while i len(tasks)任务的执行管理用from swarm import Swarm, Agentswarm.run( agentmarketing_agent, messages[{role: user, content: 帮我策划Q4的市场活动做一个完整的规划。}], context_variablescontext )4.代码检查任务1. 生成代码2. 测试代码 一系列参数exec执行。3. 大模型修正代码。4. 循环2/3直到全部成功或达到最大步数。大模型调用函数1. 用大模型输出选择调用的函数然后用正则解析获取大模型想要调用的函数名。2. 然后用字符串判定为真的方法去调用函数。大模型生成代码的执行1. 检查代码2. 用python自带的可以将字符串变为 函数名。# 执行代码提取函数 exec(code, {}, local_vars)eval执行函数并获取值。code 2 2result eval(code)AI员工需要构建一个Workspace。一种尝试围绕Agent构建「团队研发仓库」的实践一个让团队所有人都可以沉淀为「Agent可用资产」的载体。